hat.drivers.ssl

  1from ssl import *  # NOQA
  2
  3import enum
  4import pathlib
  5import ssl
  6import typing
  7
  8try:
  9    from hat.drivers.ssl import _ssl
 10
 11except ImportError:
 12    _ssl = None
 13
 14
 15class SslProtocol(enum.Enum):
 16    TLS_CLIENT = ssl.PROTOCOL_TLS_CLIENT
 17    TLS_SERVER = ssl.PROTOCOL_TLS_SERVER
 18
 19
 20class KeyUpdateType(enum.Enum):
 21    UPDATE_NOT_REQUESTED = 0
 22    UPDATE_REQUESTED = 1
 23
 24
 25def create_ssl_ctx(protocol: SslProtocol,
 26                   *,
 27                   check_hostname: bool = False,
 28                   verify_cert: bool = False,
 29                   load_default_certs: bool = True,
 30                   cert_path: pathlib.PurePath | None = None,
 31                   key_path: pathlib.PurePath | None = None,
 32                   ca_path: pathlib.PurePath | None = None,
 33                   password: str | None = None
 34                   ) -> ssl.SSLContext:
 35    ctx = ssl.SSLContext(protocol.value)
 36    ctx.check_hostname = check_hostname
 37
 38    if verify_cert:
 39        ctx.verify_mode = ssl.VerifyMode.CERT_REQUIRED
 40
 41        if load_default_certs:
 42            ctx.load_default_certs(ssl.Purpose.CLIENT_AUTH
 43                                   if protocol == SslProtocol.TLS_SERVER
 44                                   else ssl.Purpose.SERVER_AUTH)
 45
 46        if ca_path:
 47            ctx.load_verify_locations(cafile=str(ca_path))
 48
 49    else:
 50        ctx.verify_mode = ssl.VerifyMode.CERT_NONE
 51
 52    if cert_path:
 53        ctx.load_cert_chain(certfile=str(cert_path),
 54                            keyfile=str(key_path) if key_path else None,
 55                            password=password)
 56
 57    return ctx
 58
 59
 60def key_update(ssl_object: ssl.SSLObject,
 61               update_type: KeyUpdateType):
 62    if not _ssl:
 63        raise Exception('not supported')
 64
 65    if not isinstance(ssl_object, ssl.SSLObject):
 66        raise TypeError('invalid ssl object')
 67
 68    result = _ssl.key_update(ssl_object._sslobj, update_type.value)
 69    if result != 1:
 70        raise Exception('key update error')
 71
 72
 73def renegotiate(ssl_object: ssl.SSLObject):
 74    if not _ssl:
 75        raise Exception('not supported')
 76
 77    if not isinstance(ssl_object, ssl.SSLObject):
 78        raise TypeError('invalid ssl object')
 79
 80    result = _ssl.renegotiate(ssl_object._sslobj)
 81    if result != 1:
 82        raise Exception('renegotiate error')
 83
 84
 85def get_peer_cert(ssl_object: ssl.SSLObject) -> typing.Optional['Cert']:
 86    if not _ssl:
 87        raise Exception('not supported')
 88
 89    if not isinstance(ssl_object, ssl.SSLObject):
 90        raise TypeError('invalid ssl object')
 91
 92    handle = _ssl.get_peer_cert(ssl_object._sslobj)
 93    if not handle:
 94        return
 95
 96    return Cert(handle)
 97
 98
 99def load_crl(path: pathlib.PurePath) -> 'Crl':
100    if not _ssl:
101        raise Exception('not supported')
102
103    handle = _ssl.load_crl(str(path))
104    return Crl(handle)
105
106
107class Cert:
108
109    def __init__(self, handle):
110        self._handle = handle
111
112    def get_pub_key(self) -> 'PubKey':
113        handle = ssl.get_cert_pub_key(self._handle)
114        return PubKey(handle)
115
116    def get_bytes(self) -> bytes:
117        return ssl.get_cert_bytes(self._handle)
118
119
120class PubKey:
121
122    def __init__(self, handle):
123        self._handle = handle
124
125    def is_rsa(self) -> bool:
126        return ssl.is_pub_key_rsa(self._handle)
127
128    def get_size(self) -> int:
129        return ssl.get_pub_key_size(self._handle)
130
131
132class Crl:
133
134    def __init__(self, handle):
135        self._handle = handle
136
137    def contains_cert(self, cert: Cert) -> bool:
138        if not isinstance(cert, Cert):
139            raise TypeError('invalid cert')
140
141        return ssl.crl_contains_cert(self._handle, cert._handle)