ServerNameCallback=Callable[[TLSBufferObject,Optional[str],TLSConfiguration],Any]


_configuration_fields=[
'validate_certificates',
'certificate_chain',
'ciphers',
'inner_protocols',
'lowest_supported_version',
'highest_supported_version',
'trust_store',
'sni_callback',
]


_DEFAULT_VALUE=object()


classTLSConfiguration(namedtuple('TLSConfiguration',_configuration_fields)):
"""
An immutable TLS Configuration object. This object has the following
properties:

:param validate_certificates bool: Whether to validate the TLS
certificates. This switch operates at a very broad scope: either
validation is enabled, in which case all forms of validation are
performed including hostname validation if possible, or validation
is disabled, in which case no validation is performed.

Not all backends support having their certificate validation
disabled. If a backend does not support having their certificate
validation disabled, attempting to set this property to ``False``
will throw a ``TLSError`` when this object is passed into a
context object.

:param certificate_chain Tuple[Tuple[Certificate],PrivateKey]: The
certificate, intermediate certificate, and the corresponding
private key for the leaf certificate. These certificates will be
offered to the remote peer during the handshake if required.

The first Certificate in the list must be the leaf certificate. All
subsequent certificates will be offered as intermediate additional
certificates.

:param ciphers Tuple[Union[CipherSuite, int]]:
The available ciphers for TLS connections created with this
configuration, in priority order.

:param inner_protocols Tuple[Union[NextProtocol, bytes]]:
Protocols that connections created with this configuration should
advertise as supported during the TLS handshake. These may be
advertised using either or both of ALPN or NPN. This list of
protocols should be ordered by preference.

:param lowest_supported_version TLSVersion:
The minimum version of TLS that should be allowed on TLS
connections using this configuration.

:param highest_supported_version TLSVersion:
The maximum version of TLS that should be allowed on TLS
connections using this configuration.

:param trust_store TrustStore:
The trust store that connections using this configuration will use
to validate certificates.

:param sni_callback Optional[ServerNameCallback]:
A callback function that will be called after the TLS Client Hello
handshake message has been received by the TLS server when the TLS
client specifies a server name indication.

Only one callback can be set per ``TLSConfiguration``. If the
``sni_callback`` is ``None`` then the callback is disabled. If the
``TLSConfiguration`` is used for a ``ClientContext`` then this
setting will be ignored.

The ``callback`` function will be called with three arguments: the
first will be the ``TLSBufferObject`` for the connection; the
second will be a string that represents the server name that the
client is intending to communicate (or ``None`` if the TLS Client
Hello does not contain a server name); and the third argument will
be the original ``TLSConfiguration`` that configured the
connection. The server name argument will be the IDNA *decoded*
server name.

The ``callback`` must return a ``TLSConfiguration`` to allow
negotiation to continue. Other return values signal errors.
Attempting to control what error is signaled by the underlying TLS
implementation is not specified in this API, but is up to the
concrete implementation to handle.

The Context will do its best to apply the ``TLSConfiguration``
changes from its original configuration to the incoming connection.
This will usually include changing the certificate chain, but may
also include changes to allowable ciphers or any other
configuration settings.
"""
__slots__=()

def__new__(cls,validate_certificates:Optional[bool]=None,
certificate_chain:Optional[Tuple[Tuple[Certificate],PrivateKey]]=None,
ciphers:Optional[Tuple[Union[CipherSuite,int]]]=None,
inner_protocols:Optional[Tuple[Union[NextProtocol,bytes]]]=None,
lowest_supported_version:Optional[TLSVersion]=None,
highest_supported_version:Optional[TLSVersion]=None,
trust_store:Optional[TrustStore]=None,
sni_callback:Optional[ServerNameCallback]=None):

ifvalidate_certificatesisNone:
validate_certificates=True

ifciphersisNone:
ciphers=DEFAULT_CIPHER_LIST

ifinner_protocolsisNone:
inner_protocols=[]

iflowest_supported_versionisNone:
lowest_supported_version=TLSVersion.TLSv1

ifhighest_supported_versionisNone:
highest_supported_version=TLSVersion.MAXIMUM_SUPPORTED

returnsuper().__new__(
cls,validate_certificates,certificate_chain,ciphers,
inner_protocols,lowest_supported_version,
highest_supported_version,trust_store,sni_callback
)

defupdate(self,validate_certificates=_DEFAULT_VALUE,
certificate_chain=_DEFAULT_VALUE,
ciphers=_DEFAULT_VALUE,
inner_protocols=_DEFAULT_VALUE,
lowest_supported_version=_DEFAULT_VALUE,
highest_supported_version=_DEFAULT_VALUE,
trust_store=_DEFAULT_VALUE,
sni_callback=_DEFAULT_VALUE):
"""
Create a new ``TLSConfiguration``, overriding some of the settings
on the original configuration with the new settings.
"""
ifvalidate_certificatesis_DEFAULT_VALUE:
validate_certificates=self.validate_certificates

ifcertificate_chainis_DEFAULT_VALUE:
certificate_chain=self.certificate_chain

ifciphersis_DEFAULT_VALUE:
ciphers=self.ciphers

ifinner_protocolsis_DEFAULT_VALUE:
inner_protocols=self.inner_protocols

iflowest_supported_versionis_DEFAULT_VALUE:
lowest_supported_version=self.lowest_supported_version

ifhighest_supported_versionis_DEFAULT_VALUE:
highest_supported_version=self.highest_supported_version

iftrust_storeis_DEFAULT_VALUE:
trust_store=self.trust_store

ifsni_callbackis_DEFAULT_VALUE:
sni_callback=self.sni_callback

returnself.__class__(
validate_certificates,certificate_chain,ciphers,
inner_protocols,lowest_supported_version,
highest_supported_version,trust_store,sni_callback
)

TLSBufferObject=Union[TLSWrappedSocket,TLSWrappedBuffer]


class_BaseContext(metaclass=ABCMeta):
@abstractmethod
def__init__(self,configuration:TLSConfiguration):
"""
Create a new context object from a given TLS configuration.
"""

@property
@abstractmethod
defconfiguration(self)->TLSConfiguration:
"""
Returns the TLS configuration that was used to create the context.
"""


classClientContext(_BaseContext):
defwrap_socket(self,
socket:socket.socket,
server_hostname:Optional[str])->TLSWrappedSocket:
"""
Wrap an existing Python socket object ``socket`` and return a
``TLSWrappedSocket`` object. ``socket`` must be a ``SOCK_STREAM``
socket: all other socket types are unsupported.

The returned SSL socket is tied to the context, its settings and
certificates. The socket object originally passed to this method
should not be used again: attempting to use it in any way will lead
to undefined behaviour, especially across different TLS
implementations. To get the original socket object back once it has
been wrapped in TLS, see the ``unwrap`` method of the
TLSWrappedSocket.

The parameter ``server_hostname`` specifies the hostname of the
service which we are connecting to. This allows a single server to
host multiple SSL-based services with distinct certificates, quite
similarly to HTTP virtual hosts. This is also used to validate the
TLS certificate for the given hostname. If hostname validation is
not desired, then pass ``None`` for this parameter. This parameter
has no default value because opting-out of hostname validation is
dangerous, and should not be the default behaviour.
"""
buffer=self.wrap_buffers(server_hostname)
returnTLSWrappedSocket(socket,buffer)

@abstractmethod
defwrap_buffers(self,server_hostname:Optional[str])->TLSWrappedBuffer:
"""
Create an in-memory stream for TLS, using memory buffers to store
incoming and outgoing ciphertext. The TLS routines will read
received TLS data from one buffer, and write TLS data that needs to
be emitted to another buffer.

The implementation details of how this buffering works are up to
the individual TLS implementation. This allows TLS libraries that
have their own specialised support to continue to do so, while
allowing those without to use whatever Python objects they see fit.

The ``server_hostname`` parameter has the same meaning as in
``wrap_socket``.
"""


classServerContext(_BaseContext):
defwrap_socket(self,socket:socket.socket)->TLSWrappedSocket:
"""
Wrap an existing Python socket object ``socket`` and return a
``TLSWrappedSocket`` object. ``socket`` must be a ``SOCK_STREAM``
socket: all other socket types are unsupported.

The returned SSL socket is tied to the context, its settings and
certificates. The socket object originally passed to this method
should not be used again: attempting to use it in any way will lead
to undefined behaviour, especially across different TLS
implementations. To get the original socket object back once it has
been wrapped in TLS, see the ``unwrap`` method of the
TLSWrappedSocket.
"""
buffer=self.wrap_buffers()
returnTLSWrappedSocket(socket,buffer)

@abstractmethod
defwrap_buffers(self)->TLSWrappedBuffer:
"""
Create an in-memory stream for TLS, using memory buffers to store
incoming and outgoing ciphertext. The TLS routines will read
received TLS data from one buffer, and write TLS data that needs to
be emitted to another buffer.

The implementation details of how this buffering works are up to
the individual TLS implementation. This allows TLS libraries that
have their own specialised support to continue to do so, while
allowing those without to use whatever Python objects they see fit.
"""

classTLSWrappedBuffer(metaclass=ABCMeta):
@abstractmethod
defread(self,amt:int)->bytes:
"""
Read up to ``amt`` bytes of data from the input buffer and return
the result as a ``bytes`` instance.

Once EOF is reached, all further calls to this method return the
empty byte string ``b''``.

May read "short": that is, fewer bytes may be returned than were
requested.

Raise ``WantReadError`` or ``WantWriteError`` if there is
insufficient data in either the input or output buffer and the
operation would have caused data to be written or read.

May raise ``RaggedEOF`` if the connection has been closed without a
graceful TLS shutdown. Whether this is an exception that should be
ignored or not is up to the specific application.

As at any time a re-negotiation is possible, a call to ``read()``
can also cause write operations.
"""

@abstractmethod
defreadinto(self,buffer:Any,amt:int)->int:
"""
Read up to ``amt`` bytes of data from the input buffer into
``buffer``, which must be an object that implements the buffer
protocol. Returns the number of bytes read.

Once EOF is reached, all further calls to this method return the
empty byte string ``b''``.

Raises ``WantReadError`` or ``WantWriteError`` if there is
insufficient data in either the input or output buffer and the
operation would have caused data to be written or read.

May read "short": that is, fewer bytes may be read than were
requested.

May raise ``RaggedEOF`` if the connection has been closed without a
graceful TLS shutdown. Whether this is an exception that should be
ignored or not is up to the specific application.

As at any time a re-negotiation is possible, a call to
``readinto()`` can also cause write operations.
"""

@abstractmethod
defwrite(self,buf:Any)->int:
"""
Write ``buf`` in encrypted form to the output buffer and return the
number of bytes written. The ``buf`` argument must be an object
supporting the buffer interface.

Raise ``WantReadError`` or ``WantWriteError`` if there is
insufficient data in either the input or output buffer and the
operation would have caused data to be written or read. In either
case, users should endeavour to resolve that situation and then
re-call this method. When re-calling this method users *should*
re-use the exact same ``buf`` object, as some backends require that
the exact same buffer be used.

This operation may write "short": that is, fewer bytes may be
written than were in the buffer.

As at any time a re-negotiation is possible, a call to ``write()``
can also cause read operations.
"""

@abstractmethod
defdo_handshake(self)->None:
"""
Performs the TLS handshake. Also performs certificate validation
and hostname verification.
"""

@abstractmethod
defcipher(self)->Optional[Union[CipherSuite,int]]:
"""
Returns the CipherSuite entry for the cipher that has been
negotiated on the connection. If no connection has been negotiated,
returns ``None``. If the cipher negotiated is not defined in
CipherSuite, returns the 16-bit integer representing that cipher
directly.
"""

@abstractmethod
defnegotiated_protocol(self)->Optional[Union[NextProtocol,bytes]]:
"""
Returns the protocol that was selected during the TLS handshake.
This selection may have been made using ALPN, NPN, or some future
negotiation mechanism.

If the negotiated protocol is one of the protocols defined in the
``NextProtocol`` enum, the value from that enum will be returned.
Otherwise, the raw bytestring of the negotiated protocol will be
returned.

If ``Context.set_inner_protocols()`` was not called, if the other
party does not support protocol negotiation, if this socket does
not support any of the peer's proposed protocols, or if the
handshake has not happened yet, ``None`` is returned.
"""

@property
@abstractmethod
defcontext(self)->Context:
"""
The ``Context`` object this buffer is tied to.
"""

@abstractproperty
defnegotiated_tls_version(self)->Optional[TLSVersion]:
"""
The version of TLS that has been negotiated on this connection.
"""

@abstractmethod
defshutdown(self)->None:
"""
Performs a clean TLS shut down. This should generally be used
whenever possible to signal to the remote peer that the content is
finished.
"""

@abstractmethod
defreceive_from_network(self,data):
"""
Receives some TLS data from the network and stores it in an
internal buffer.
"""

@abstractmethod
defpeek_outgoing(self,amt):
"""
Returns the next ``amt`` bytes of data that should be written to
the network from the outgoing data buffer, without removing it from
the internal buffer.
"""

@abstractmethod
defconsume_outgoing(self,amt):
"""
Discard the next ``amt`` bytes from the outgoing data buffer. This
should be used when ``amt`` bytes have been sent on the network, to
signal that the data no longer needs to be buffered.
"""

classTLSWrappedSocket:
defdo_handshake(self)->None:
"""
Performs the TLS handshake. Also performs certificate validation
and hostname verification. This must be called after the socket has
connected (either via ``connect`` or ``accept``), before any other
operation is performed on the socket.
"""

defcipher(self)->Optional[Union[CipherSuite,int]]:
"""
Returns the CipherSuite entry for the cipher that has been
negotiated on the connection. If no connection has been negotiated,
returns ``None``. If the cipher negotiated is not defined in
CipherSuite, returns the 16-bit integer representing that cipher
directly.
"""

defnegotiated_protocol(self)->Optional[Union[NextProtocol,bytes]]:
"""
Returns the protocol that was selected during the TLS handshake.
This selection may have been made using ALPN, NPN, or some future
negotiation mechanism.

If the negotiated protocol is one of the protocols defined in the
``NextProtocol`` enum, the value from that enum will be returned.
Otherwise, the raw bytestring of the negotiated protocol will be
returned.

If ``Context.set_inner_protocols()`` was not called, if the other
party does not support protocol negotiation, if this socket does
not support any of the peer's proposed protocols, or if the
handshake has not happened yet, ``None`` is returned.
"""

@property
defcontext(self)->Context:
"""
The ``Context`` object this socket is tied to.
"""

defnegotiated_tls_version(self)->Optional[TLSVersion]:
"""
The version of TLS that has been negotiated on this connection.
"""

defunwrap(self)->socket.socket:
"""
Cleanly terminate the TLS connection on this wrapped socket. Once
called, this ``TLSWrappedSocket`` can no longer be used to transmit
data. Returns the socket that was wrapped with TLS.
"""

classNextProtocol(Enum):
H2=b'h2'
H2C=b'h2c'
HTTP1=b'http/1.1'
WEBRTC=b'webrtc'
C_WEBRTC=b'c-webrtc'
FTP=b'ftp'
STUN=b'stun.nat-discovery'
TURN=b'stun.turn'

classTLSVersion(Enum):
MINIMUM_SUPPORTED=auto()
SSLv2=auto()
SSLv3=auto()
TLSv1=auto()
TLSv1_1=auto()
TLSv1_2=auto()
TLSv1_3=auto()
MAXIMUM_SUPPORTED=auto()

classTLSError(Exception):
"""
The base exception for all TLS related errors from any backend.
Catching this error should be sufficient to catch *all* TLS errors,
regardless of what backend is used.
"""

classWantWriteError(TLSError):
"""
A special signaling exception used only when non-blocking or
buffer-only I/O is used. This error signals that the requested
operation cannot complete until more data is written to the network,
or until the output buffer is drained.

This error is should only be raised when it is completely impossible
to write any data. If a partial write is achievable then this should
not be raised.
"""

classWantReadError(TLSError):
"""
A special signaling exception used only when non-blocking or
buffer-only I/O is used. This error signals that the requested
operation cannot complete until more data is read from the network, or
until more data is available in the input buffer.

This error should only be raised when it is completely impossible to
write any data. If a partial write is achievable then this should not
be raised.
"""

classRaggedEOF(TLSError):
"""
A special signaling exception used when a TLS connection has been
closed gracelessly: that is, when a TLS CloseNotify was not received
from the peer before the underlying TCP socket reached EOF. This is a
so-called "ragged EOF".

This exception is not guaranteed to be raised in the face of a ragged
EOF: some implementations may not be able to detect or report the
ragged EOF.

This exception is not always a problem. Ragged EOFs are a concern only
when protocols are vulnerable to length truncation attacks. Any
protocol that can detect length truncation attacks at the application
layer (e.g. HTTP/1.1 and HTTP/2) is not vulnerable to this kind of
attack and so can ignore this exception.
"""

classCertificate(metaclass=ABCMeta):
@abstractclassmethod
deffrom_buffer(cls,buffer:bytes):
"""
Creates a Certificate object from a byte buffer. This byte buffer
may be either PEM-encoded or DER-encoded. If the buffer is PEM
encoded it *must* begin with the standard PEM preamble (a series of
dashes followed by the ASCII bytes "BEGIN CERTIFICATE" and another
series of dashes). In the absence of that preamble, the
implementation may assume that the certificate is DER-encoded
instead.
"""

@abstractclassmethod
deffrom_file(cls,path:Union[pathlib.Path,AnyStr]):
"""
Creates a Certificate object from a file on disk. This method may
be a convenience method that wraps ``open`` and ``from_buffer``,
but some TLS implementations may be able to provide more-secure or
faster methods of loading certificates that do not involve Python
code.
"""

classPrivateKey(metaclass=ABCMeta):
@abstractclassmethod
deffrom_buffer(cls,
buffer:bytes,
password:Optional[Union[Callable[[],Union[bytes,bytearray]],bytes,bytearray]]=None):
"""
Creates a PrivateKey object from a byte buffer. This byte buffer
may be either PEM-encoded or DER-encoded. If the buffer is PEM
encoded it *must* begin with the standard PEM preamble (a series of
dashes followed by the ASCII bytes "BEGIN", the key type, and
another series of dashes). In the absence of that preamble, the
implementation may assume that the certificate is DER-encoded
instead.

The key may additionally be encrypted. If it is, the ``password``
argument can be used to decrypt the key. The ``password`` argument
may be a function to call to get the password for decrypting the
private key. It will only be called if the private key is encrypted
and a password is necessary. It will be called with no arguments,
and it should return either bytes or bytearray containing the
password. Alternatively a bytes, or bytearray value may be supplied
directly as the password argument. It will be ignored if the
private key is not encrypted and no password is needed.
"""

@abstractclassmethod
deffrom_file(cls,
path:Union[pathlib.Path,bytes,str],
password:Optional[Union[Callable[[],Union[bytes,bytearray]],bytes,bytearray]]=None):
"""
Creates a PrivateKey object from a file on disk. This method may
be a convenience method that wraps ``open`` and ``from_buffer``,
but some TLS implementations may be able to provide more-secure or
faster methods of loading certificates that do not involve Python
code.

The ``password`` parameter behaves exactly as the equivalent
parameter on ``from_buffer``.
"""

classTrustStore(metaclass=ABCMeta):
@abstractclassmethod
defsystem(cls)->TrustStore:
"""
Returns a TrustStore object that represents the system trust
database.
"""

@abstractclassmethod
deffrom_pem_file(cls,path:Union[pathlib.Path,bytes,str])->TrustStore:
"""
Initializes a trust store from a single file full of PEMs.
"""

Backend=namedtuple(
'Backend',
['client_context','server_context',
'certificate','private_key','trust_store']
)

trust_store=backend.trust_store.system()