text: import datetime
  import logging
  import os
  import ssl
  import struct
  from contextlib import contextmanager
  from dataclasses import dataclass, field
  from enum import Enum, IntEnum
  from functools import partial
  from typing import (
      Any,
      Callable,
      Dict,
      Generator,
      List,
      Optional,
      Sequence,
      Tuple,
      TypeVar,
      Union,
  )
  
  import certifi
  from cryptography import x509
  from cryptography.exceptions import InvalidSignature
  from cryptography.hazmat.backends import default_backend
  from cryptography.hazmat.bindings.openssl.binding import Binding
  from cryptography.hazmat.primitives import hashes, hmac, serialization
  from cryptography.hazmat.primitives.asymmetric import (
      dsa,
      ec,
      padding,
      rsa,
      x448,
      x25519,
      ed25519,
  )
  from cryptography.hazmat.primitives.kdf.hkdf import HKDFExpand
  from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
  
  from .buffer import Buffer
  
  binding = Binding()
  binding.init_static_locks()
  ffi = binding.ffi
  lib = binding.lib
  
  TLS_VERSION_1_2 = 0x0303
  TLS_VERSION_1_3 = 0x0304
  TLS_VERSION_1_3_DRAFT_28 = 0x7F1C
  TLS_VERSION_1_3_DRAFT_27 = 0x7F1B
  TLS_VERSION_1_3_DRAFT_26 = 0x7F1A
  
  T = TypeVar("T")
  
  # facilitate mocking for the test suite
  utcnow = datetime.datetime.utcnow
  
  
  class AlertDescription(IntEnum):
      close_notify = 0
      unexpected_message = 10
      bad_record_mac = 20
      record_overflow = 22
      handshake_failure = 40
      bad_certificate = 42
      unsupported_certificate = 43
      certificate_revoked = 44
      certificate_expired = 45
      certificate_unknown = 46
      illegal_parameter = 47
      unknown_ca = 48
      access_denied = 49
      decode_error = 50
      decrypt_error = 51
      protocol_version = 70
      insufficient_security = 71
      internal_error = 80
      inappropriate_fallback = 86
      user_canceled = 90
      missing_extension = 109
      unsupported_extension = 110
      unrecognized_name = 112
      bad_certificate_status_response = 113
      unknown_psk_identity = 115
      certificate_required = 116
      no_application_protocol = 120
  
  
  class Alert(Exception):
      description: AlertDescription
  
  
  class AlertBadCertificate(Alert):
      description = AlertDescription.bad_certificate
  
  
  class AlertCertificateExpired(Alert):
      description = AlertDescription.certificate_expired
  
  
  class AlertDecryptError(Alert):
      description = AlertDescription.decrypt_error
  
  
  class AlertHandshakeFailure(Alert):
      description = AlertDescription.handshake_failure
  
  
  class AlertIllegalParameter(Alert):
      description = AlertDescription.illegal_parameter
  
  
  class AlertInternalError(Alert):
      description = AlertDescription.internal_error
  
  
  class AlertProtocolVersion(Alert):
      description = AlertDescription.protocol_version
  
  
  class AlertUnexpectedMessage(Alert):
      description = AlertDescription.unexpected_message
  
  
  class Direction(Enum):
      DECRYPT = 0
      ENCRYPT = 1
  
  
  class Epoch(Enum):
      INITIAL = 0
      ZERO_RTT = 1
      HANDSHAKE = 2
      ONE_RTT = 3
  
  
  class State(Enum):
      CLIENT_HANDSHAKE_START = 0
      CLIENT_EXPECT_SERVER_HELLO = 1
      CLIENT_EXPECT_ENCRYPTED_EXTENSIONS = 2
      CLIENT_EXPECT_CERTIFICATE_REQUEST_OR_CERTIFICATE = 3
      CLIENT_EXPECT_CERTIFICATE_CERTIFICATE = 4
      CLIENT_EXPECT_CERTIFICATE_VERIFY = 5
      CLIENT_EXPECT_FINISHED = 6
      CLIENT_POST_HANDSHAKE = 7
  
      SERVER_EXPECT_CLIENT_HELLO = 8
      SERVER_EXPECT_FINISHED = 9
      SERVER_POST_HANDSHAKE = 10
  
  
  def hkdf_label(label: bytes, hash_value: bytes, length: int) -> bytes:
      full_label = b"tls13 " + label
      return (
          struct.pack("!HB", length, len(full_label))
          + full_label
          + struct.pack("!B", len(hash_value))
          + hash_value
      )
  
  
  def hkdf_expand_label(
      algorithm: hashes.HashAlgorithm,
      secret: bytes,
      label: bytes,
      hash_value: bytes,
      length: int,
  ) -> bytes:
      return HKDFExpand(
          algorithm=algorithm,
          length=length,
          info=hkdf_label(label, hash_value, length),
          backend=default_backend(),
      ).derive(secret)
  
  
  def hkdf_extract(
      algorithm: hashes.HashAlgorithm, salt: bytes, key_material: bytes
  ) -> bytes:
      h = hmac.HMAC(salt, algorithm, backend=default_backend())
      h.update(key_material)
      return h.finalize()
  
  
  def load_pem_private_key(
      data: bytes, password: Optional[bytes] = None
  ) -> Union[dsa.DSAPrivateKey, ec.EllipticCurvePrivateKey, rsa.RSAPrivateKey]:
      """
      Load a PEM-encoded private key.
      """
      return serialization.load_pem_private_key(
          data, password=password, backend=default_backend()
      )
  
  
  def load_pem_x509_certificates(data: bytes) -> List[x509.Certificate]:
      """
      Load a chain of PEM-encoded X509 certificates.
      """
      boundary = b"-----END CERTIFICATE-----\n"
      certificates = []
      for chunk in data.split(boundary):
          if chunk:
              certificates.append(
                  x509.load_pem_x509_certificate(
                      chunk + boundary, backend=default_backend()
                  )
              )
      return certificates
  
  
  def openssl_assert(ok: bool, func: str) -> None:
      if not ok:
          lib.ERR_clear_error()
          raise AlertInternalError("OpenSSL call to %s failed" % func)
  
  
  def openssl_decode_string(charp) -> str:
      return ffi.string(charp).decode("utf-8") if charp else ""
  
  
  def openssl_encode_path(s: Optional[str]) -> Any:
      if s is not None:
          return os.fsencode(s)
      return ffi.NULL
  
  
  def cert_x509_ptr(certificate: x509.Certificate) -> Any:
      """
      Accessor for private attribute.
      """
      return getattr(certificate, "_x509")
  
  
  def verify_certificate(
      certificate: x509.Certificate,
      chain: List[x509.Certificate] = [],
      server_name: Optional[str] = None,
      cadata: Optional[bytes] = None,
      cafile: Optional[str] = None,
      capath: Optional[str] = None,
  ) -> None:
      # verify dates
      now = utcnow()
      if now < certificate.not_valid_before:
          raise AlertCertificateExpired("Certificate is not valid yet")
      if now > certificate.not_valid_after:
          raise AlertCertificateExpired("Certificate is no longer valid")
  
      # verify subject
      if server_name is not None:
          subject = []
          subjectAltName: List[Tuple[str, str]] = []
          for attr in certificate.subject:
              if attr.oid == x509.NameOID.COMMON_NAME:
                  subject.append((("commonName", attr.value),))
          for ext in certificate.extensions:
              if isinstance(ext.value, x509.SubjectAlternativeName):
                  for name in ext.value:
                      if isinstance(name, x509.DNSName):
                          subjectAltName.append(("DNS", name.value))
  
          try:
              ssl.match_hostname(
                  {"subject": tuple(subject), "subjectAltName": tuple(subjectAltName)},
                  server_name,
              )
          except ssl.CertificateError as exc:
              raise AlertBadCertificate("\n".join(exc.args)) from exc
  
      # verify certificate chain
      store = lib.X509_STORE_new()
      openssl_assert(store != ffi.NULL, "X509_store_new")
      store = ffi.gc(store, lib.X509_STORE_free)
  
      # load default CAs
      openssl_assert(
          lib.X509_STORE_set_default_paths(store), "X509_STORE_set_default_paths"
      )
      openssl_assert(
          lib.X509_STORE_load_locations(
              store,
              openssl_encode_path(certifi.where()),
              openssl_encode_path(None),
          ),
          "X509_STORE_load_locations",
      )
  
      # load extra CAs
      if cadata is not None:
          for cert in load_pem_x509_certificates(cadata):
              openssl_assert(
                  lib.X509_STORE_add_cert(store, cert_x509_ptr(cert)),
                  "X509_STORE_add_cert",
              )
  
      if cafile is not None or capath is not None:
          openssl_assert(
              lib.X509_STORE_load_locations(
                  store, openssl_encode_path(cafile), openssl_encode_path(capath)
              ),
              "X509_STORE_load_locations",
          )
  
      chain_stack = lib.sk_X509_new_null()
      openssl_assert(chain_stack != ffi.NULL, "sk_X509_new_null")
      chain_stack = ffi.gc(chain_stack, lib.sk_X509_free)
      for cert in chain:
          openssl_assert(
              lib.sk_X509_push(chain_stack, cert_x509_ptr(cert)), "sk_X509_push"
          )
  
      store_ctx = lib.X509_STORE_CTX_new()
      openssl_assert(store_ctx != ffi.NULL, "X509_STORE_CTX_new")
      store_ctx = ffi.gc(store_ctx, lib.X509_STORE_CTX_free)
      openssl_assert(
          lib.X509_STORE_CTX_init(
              store_ctx, store, cert_x509_ptr(certificate), chain_stack
          ),
          "X509_STORE_CTX_init",
      )
  
      res = lib.X509_verify_cert(store_ctx)
      if not res:
          err = lib.X509_STORE_CTX_get_error(store_ctx)
          err_str = openssl_decode_string(lib.X509_verify_cert_error_string(err))
          raise AlertBadCertificate(err_str)
  
  
  class CipherSuite(IntEnum):
      AES_128_GCM_SHA256 = 0x1301
      AES_256_GCM_SHA384 = 0x1302
      CHACHA20_POLY1305_SHA256 = 0x1303
      EMPTY_RENEGOTIATION_INFO_SCSV = 0x00FF
  
  
  class CompressionMethod(IntEnum):
      NULL = 0
  
  
  class ExtensionType(IntEnum):
      SERVER_NAME = 0
      STATUS_REQUEST = 5
      SUPPORTED_GROUPS = 10
      SIGNATURE_ALGORITHMS = 13
      ALPN = 16
      COMPRESS_CERTIFICATE = 27
      PRE_SHARED_KEY = 41
      EARLY_DATA = 42
      SUPPORTED_VERSIONS = 43
      COOKIE = 44
      PSK_KEY_EXCHANGE_MODES = 45
      KEY_SHARE = 51
      QUIC_TRANSPORT_PARAMETERS = 0x0039
      QUIC_TRANSPORT_PARAMETERS_DRAFT = 0xFFA5
      ENCRYPTED_SERVER_NAME = 65486
  
  
  class Group(IntEnum):
      SECP256R1 = 0x0017
      SECP384R1 = 0x0018
      SECP521R1 = 0x0019
      X25519 = 0x001D
      X448 = 0x001E
      GREASE = 0xAAAA
  
  
  class HandshakeType(IntEnum):
      CLIENT_HELLO = 1
      SERVER_HELLO = 2
      NEW_SESSION_TICKET = 4
      END_OF_EARLY_DATA = 5
      ENCRYPTED_EXTENSIONS = 8
      CERTIFICATE = 11
      CERTIFICATE_REQUEST = 13
      CERTIFICATE_VERIFY = 15
      FINISHED = 20
      KEY_UPDATE = 24
      COMPRESSED_CERTIFICATE = 25
      MESSAGE_HASH = 254
  
  
  class PskKeyExchangeMode(IntEnum):
      PSK_KE = 0
      PSK_DHE_KE = 1
  
  
  class SignatureAlgorithm(IntEnum):
      ECDSA_SECP256R1_SHA256 = 0x0403
      ECDSA_SECP384R1_SHA384 = 0x0503
      ECDSA_SECP521R1_SHA512 = 0x0603
      ED25519 = 0x0807
      ED448 = 0x0808
      RSA_PKCS1_SHA256 = 0x0401
      RSA_PKCS1_SHA384 = 0x0501
      RSA_PKCS1_SHA512 = 0x0601
      RSA_PSS_PSS_SHA256 = 0x0809
      RSA_PSS_PSS_SHA384 = 0x080A
      RSA_PSS_PSS_SHA512 = 0x080B
      RSA_PSS_RSAE_SHA256 = 0x0804
      RSA_PSS_RSAE_SHA384 = 0x0805
      RSA_PSS_RSAE_SHA512 = 0x0806
  
      # legacy
      RSA_PKCS1_SHA1 = 0x0201
      SHA1_DSA = 0x0202
      ECDSA_SHA1 = 0x0203
  
  
  # BLOCKS
  
  
  @contextmanager
  def pull_block(buf: Buffer, capacity: int) -> Generator:
      length = 0
      for b in buf.pull_bytes(capacity):
          length = (length << 8) | b
      end = buf.tell() + length
      yield length
      assert buf.tell() == end
  
  
  @contextmanager
  def push_block(buf: Buffer, capacity: int) -> Generator:
      """
      Context manager to push a variable-length block, with `capacity` bytes
      to write the length.
      """
      start = buf.tell() + capacity
      buf.seek(start)
      yield
      end = buf.tell()
      length = end - start
      while capacity:
          buf.seek(start - capacity)
          buf.push_uint8((length >> (8 * (capacity - 1))) & 0xFF)
          capacity -= 1
      buf.seek(end)
  
  
  # LISTS
  
  
  def pull_list(buf: Buffer, capacity: int, func: Callable[[], T]) -> List[T]:
      """
      Pull a list of items.
      """
      items = []
      with pull_block(buf, capacity) as length:
          end = buf.tell() + length
          while buf.tell() < end:
              items.append(func())
      return items
  
  
  def push_list(
      buf: Buffer, capacity: int, func: Callable[[T], None], values: Sequence[T]
  ) -> None:
      """
      Push a list of items.
      """
      with push_block(buf, capacity):
          for value in values:
              func(value)
  
  
  def pull_opaque(buf: Buffer, capacity: int) -> bytes:
      """
      Pull an opaque value prefixed by a length.
      """
      with pull_block(buf, capacity) as length:
          return buf.pull_bytes(length)
  
  
  def push_opaque(buf: Buffer, capacity: int, value: bytes) -> None:
      """
      Push an opaque value prefix by a length.
      """
      with push_block(buf, capacity):
          buf.push_bytes(value)
  
  
  @contextmanager
  def push_extension(buf: Buffer, extension_type: int) -> Generator:
      buf.push_uint16(extension_type)
      with push_block(buf, 2):
          yield
  
  
  # KeyShareEntry
  
  
  KeyShareEntry = Tuple[int, bytes]
  
  
  def pull_key_share(buf: Buffer) -> KeyShareEntry:
      group = buf.pull_uint16()
      data = pull_opaque(buf, 2)
      return (group, data)
  
  
  def push_key_share(buf: Buffer, value: KeyShareEntry) -> None:
      buf.push_uint16(value[0])
      push_opaque(buf, 2, value[1])
  
  
  # ALPN
  
  
  def pull_alpn_protocol(buf: Buffer) -> str:
      return pull_opaque(buf, 1).decode("ascii")
  
  
  def push_alpn_protocol(buf: Buffer, protocol: str) -> None:
      push_opaque(buf, 1, protocol.encode("ascii"))
  
  
  # PRE SHARED KEY
  
  PskIdentity = Tuple[bytes, int]
  
  
  def pull_psk_identity(buf: Buffer) -> PskIdentity:
      identity = pull_opaque(buf, 2)
      obfuscated_ticket_age = buf.pull_uint32()
      return (identity, obfuscated_ticket_age)
  
  
  def push_psk_identity(buf: Buffer, entry: PskIdentity) -> None:
      push_opaque(buf, 2, entry[0])
      buf.push_uint32(entry[1])
  
  
  def pull_psk_binder(buf: Buffer) -> bytes:
      return pull_opaque(buf, 1)
  
  
  def push_psk_binder(buf: Buffer, binder: bytes) -> None:
      push_opaque(buf, 1, binder)
  
  
  # MESSAGES
  
  Extension = Tuple[int, bytes]
  
  
  @dataclass
  class OfferedPsks:
      identities: List[PskIdentity]
      binders: List[bytes]
  
  
  @dataclass
  class ClientHello:
      random: bytes
      legacy_session_id: bytes
      cipher_suites: List[int]
      legacy_compression_methods: List[int]
  
      # extensions
      alpn_protocols: Optional[List[str]] = None
      early_data: bool = False
      key_share: Optional[List[KeyShareEntry]] = None
      pre_shared_key: Optional[OfferedPsks] = None
      psk_key_exchange_modes: Optional[List[int]] = None
      server_name: Optional[str] = None
      signature_algorithms: Optional[List[int]] = None
      supported_groups: Optional[List[int]] = None
      supported_versions: Optional[List[int]] = None
  
      other_extensions: List[Extension] = field(default_factory=list)
  
  
  def pull_client_hello(buf: Buffer) -> ClientHello:
      assert buf.pull_uint8() == HandshakeType.CLIENT_HELLO
      with pull_block(buf, 3):
          assert buf.pull_uint16() == TLS_VERSION_1_2
  
          hello = ClientHello(
              random=buf.pull_bytes(32),
              legacy_session_id=pull_opaque(buf, 1),
              cipher_suites=pull_list(buf, 2, buf.pull_uint16),
              legacy_compression_methods=pull_list(buf, 1, buf.pull_uint8),
          )
  
          # extensions
          after_psk = False
  
          def pull_extension() -> None:
              # pre_shared_key MUST be last
              nonlocal after_psk
              assert not after_psk
  
              extension_type = buf.pull_uint16()
              extension_length = buf.pull_uint16()
              if extension_type == ExtensionType.KEY_SHARE:
                  hello.key_share = pull_list(buf, 2, partial(pull_key_share, buf))
              elif extension_type == ExtensionType.SUPPORTED_VERSIONS:
                  hello.supported_versions = pull_list(buf, 1, buf.pull_uint16)
              elif extension_type == ExtensionType.SIGNATURE_ALGORITHMS:
                  hello.signature_algorithms = pull_list(buf, 2, buf.pull_uint16)
              elif extension_type == ExtensionType.SUPPORTED_GROUPS:
                  hello.supported_groups = pull_list(buf, 2, buf.pull_uint16)
              elif extension_type == ExtensionType.PSK_KEY_EXCHANGE_MODES:
                  hello.psk_key_exchange_modes = pull_list(buf, 1, buf.pull_uint8)
              elif extension_type == ExtensionType.SERVER_NAME:
                  with pull_block(buf, 2):
                      assert buf.pull_uint8() == 0
                      hello.server_name = pull_opaque(buf, 2).decode("ascii")
              elif extension_type == ExtensionType.ALPN:
                  hello.alpn_protocols = pull_list(
                      buf, 2, partial(pull_alpn_protocol, buf)
                  )
              elif extension_type == ExtensionType.EARLY_DATA:
                  hello.early_data = True
              elif extension_type == ExtensionType.PRE_SHARED_KEY:
                  hello.pre_shared_key = OfferedPsks(
                      identities=pull_list(buf, 2, partial(pull_psk_identity, buf)),
                      binders=pull_list(buf, 2, partial(pull_psk_binder, buf)),
                  )
                  after_psk = True
              else:
                  hello.other_extensions.append(
                      (extension_type, buf.pull_bytes(extension_length))
                  )
  
          pull_list(buf, 2, pull_extension)
  
      return hello
  
  
  def push_client_hello(buf: Buffer, hello: ClientHello) -> None:
      buf.push_uint8(HandshakeType.CLIENT_HELLO)
      with push_block(buf, 3):
          buf.push_uint16(TLS_VERSION_1_2)
          buf.push_bytes(hello.random)
          push_opaque(buf, 1, hello.legacy_session_id)
          push_list(buf, 2, buf.push_uint16, hello.cipher_suites)
          push_list(buf, 1, buf.push_uint8, hello.legacy_compression_methods)
  
          # extensions
          with push_block(buf, 2):
              with push_extension(buf, ExtensionType.KEY_SHARE):
                  push_list(buf, 2, partial(push_key_share, buf), hello.key_share)
  
              with push_extension(buf, ExtensionType.SUPPORTED_VERSIONS):
                  push_list(buf, 1, buf.push_uint16, hello.supported_versions)
  
              with push_extension(buf, ExtensionType.SIGNATURE_ALGORITHMS):
                  push_list(buf, 2, buf.push_uint16, hello.signature_algorithms)
  
              with push_extension(buf, ExtensionType.SUPPORTED_GROUPS):
                  push_list(buf, 2, buf.push_uint16, hello.supported_groups)
  
              if hello.psk_key_exchange_modes is not None:
                  with push_extension(buf, ExtensionType.PSK_KEY_EXCHANGE_MODES):
                      push_list(buf, 1, buf.push_uint8, hello.psk_key_exchange_modes)
  
              if hello.server_name is not None:
                  with push_extension(buf, ExtensionType.SERVER_NAME):
                      with push_block(buf, 2):
                          buf.push_uint8(0)
                          push_opaque(buf, 2, hello.server_name.encode("ascii"))
  
              if hello.alpn_protocols is not None:
                  with push_extension(buf, ExtensionType.ALPN):
                      push_list(
                          buf, 2, partial(push_alpn_protocol, buf), hello.alpn_protocols
                      )
  
              for extension_type, extension_value in hello.other_extensions:
                  with push_extension(buf, extension_type):
                      buf.push_bytes(extension_value)
  
              if hello.early_data:
                  with push_extension(buf, ExtensionType.EARLY_DATA):
                      pass
  
              # pre_shared_key MUST be last
              if hello.pre_shared_key is not None:
                  with push_extension(buf, ExtensionType.PRE_SHARED_KEY):
                      push_list(
                          buf,
                          2,
                          partial(push_psk_identity, buf),
                          hello.pre_shared_key.identities,
                      )
                      push_list(
                          buf,
                          2,
                          partial(push_psk_binder, buf),
                          hello.pre_shared_key.binders,
                      )
  
  
  @dataclass
  class ServerHello:
      random: bytes
      legacy_session_id: bytes
      cipher_suite: int
      compression_method: int
  
      # extensions
      key_share: Optional[KeyShareEntry] = None
      pre_shared_key: Optional[int] = None
      supported_version: Optional[int] = None
      other_extensions: List[Tuple[int, bytes]] = field(default_factory=list)
  
  
  def pull_server_hello(buf: Buffer) -> ServerHello:
      assert buf.pull_uint8() == HandshakeType.SERVER_HELLO
      with pull_block(buf, 3):
          assert buf.pull_uint16() == TLS_VERSION_1_2
  
          hello = ServerHello(
              random=buf.pull_bytes(32),
              legacy_session_id=pull_opaque(buf, 1),
              cipher_suite=buf.pull_uint16(),
              compression_method=buf.pull_uint8(),
          )
  
          # extensions
          def pull_extension() -> None:
              extension_type = buf.pull_uint16()
              extension_length = buf.pull_uint16()
              if extension_type == ExtensionType.SUPPORTED_VERSIONS:
                  hello.supported_version = buf.pull_uint16()
              elif extension_type == ExtensionType.KEY_SHARE:
                  hello.key_share = pull_key_share(buf)
              elif extension_type == ExtensionType.PRE_SHARED_KEY:
                  hello.pre_shared_key = buf.pull_uint16()
              else:
                  hello.other_extensions.append(
                      (extension_type, buf.pull_bytes(extension_length))
                  )
  
          pull_list(buf, 2, pull_extension)
  
      return hello
  
  
  def push_server_hello(buf: Buffer, hello: ServerHello) -> None:
      buf.push_uint8(HandshakeType.SERVER_HELLO)
      with push_block(buf, 3):
          buf.push_uint16(TLS_VERSION_1_2)
          buf.push_bytes(hello.random)
  
          push_opaque(buf, 1, hello.legacy_session_id)
          buf.push_uint16(hello.cipher_suite)
          buf.push_uint8(hello.compression_method)
  
          # extensions
          with push_block(buf, 2):
              if hello.supported_version is not None:
                  with push_extension(buf, ExtensionType.SUPPORTED_VERSIONS):
                      buf.push_uint16(hello.supported_version)
  
              if hello.key_share is not None:
                  with push_extension(buf, ExtensionType.KEY_SHARE):
                      push_key_share(buf, hello.key_share)
  
              if hello.pre_shared_key is not None:
                  with push_extension(buf, ExtensionType.PRE_SHARED_KEY):
                      buf.push_uint16(hello.pre_shared_key)
  
              for extension_type, extension_value in hello.other_extensions:
                  with push_extension(buf, extension_type):
                      buf.push_bytes(extension_value)
  
  
  @dataclass
  class NewSessionTicket:
      ticket_lifetime: int = 0
      ticket_age_add: int = 0
      ticket_nonce: bytes = b""
      ticket: bytes = b""
  
      # extensions
      max_early_data_size: Optional[int] = None
      other_extensions: List[Tuple[int, bytes]] = field(default_factory=list)
  
  
  def pull_new_session_ticket(buf: Buffer) -> NewSessionTicket:
      new_session_ticket = NewSessionTicket()
  
      assert buf.pull_uint8() == HandshakeType.NEW_SESSION_TICKET
      with pull_block(buf, 3):
          new_session_ticket.ticket_lifetime = buf.pull_uint32()
          new_session_ticket.ticket_age_add = buf.pull_uint32()
          new_session_ticket.ticket_nonce = pull_opaque(buf, 1)
          new_session_ticket.ticket = pull_opaque(buf, 2)
  
          def pull_extension() -> None:
              extension_type = buf.pull_uint16()
              extension_length = buf.pull_uint16()
              if extension_type == ExtensionType.EARLY_DATA:
                  new_session_ticket.max_early_data_size = buf.pull_uint32()
              else:
                  new_session_ticket.other_extensions.append(
                      (extension_type, buf.pull_bytes(extension_length))
                  )
  
          pull_list(buf, 2, pull_extension)
  
      return new_session_ticket
  
  
  def push_new_session_ticket(buf: Buffer, new_session_ticket: NewSessionTicket) -> None:
      buf.push_uint8(HandshakeType.NEW_SESSION_TICKET)
      with push_block(buf, 3):
          buf.push_uint32(new_session_ticket.ticket_lifetime)
          buf.push_uint32(new_session_ticket.ticket_age_add)
          push_opaque(buf, 1, new_session_ticket.ticket_nonce)
          push_opaque(buf, 2, new_session_ticket.ticket)
  
          with push_block(buf, 2):
              if new_session_ticket.max_early_data_size is not None:
                  with push_extension(buf, ExtensionType.EARLY_DATA):
                      buf.push_uint32(new_session_ticket.max_early_data_size)
  
              for extension_type, extension_value in new_session_ticket.other_extensions:
                  with push_extension(buf, extension_type):
                      buf.push_bytes(extension_value)
  
  
  @dataclass
  class EncryptedExtensions:
      alpn_protocol: Optional[str] = None
      early_data: bool = False
  
      other_extensions: List[Tuple[int, bytes]] = field(default_factory=list)
  
  
  def pull_encrypted_extensions(buf: Buffer) -> EncryptedExtensions:
      extensions = EncryptedExtensions()
  
      assert buf.pull_uint8() == HandshakeType.ENCRYPTED_EXTENSIONS
      with pull_block(buf, 3):
  
          def pull_extension() -> None:
              extension_type = buf.pull_uint16()
              extension_length = buf.pull_uint16()
              if extension_type == ExtensionType.ALPN:
                  extensions.alpn_protocol = pull_list(
                      buf, 2, partial(pull_alpn_protocol, buf)
                  )[0]
              elif extension_type == ExtensionType.EARLY_DATA:
                  extensions.early_data = True
              else:
                  extensions.other_extensions.append(
                      (extension_type, buf.pull_bytes(extension_length))
                  )
  
          pull_list(buf, 2, pull_extension)
  
      return extensions
  
  
  def push_encrypted_extensions(buf: Buffer, extensions: EncryptedExtensions) -> None:
      buf.push_uint8(HandshakeType.ENCRYPTED_EXTENSIONS)
      with push_block(buf, 3):
          with push_block(buf, 2):
              if extensions.alpn_protocol is not None:
                  with push_extension(buf, ExtensionType.ALPN):
                      push_list(
                          buf,
                          2,
                          partial(push_alpn_protocol, buf),
                          [extensions.alpn_protocol],
                      )
  
              if extensions.early_data:
                  with push_extension(buf, ExtensionType.EARLY_DATA):
                      pass
  
              for extension_type, extension_value in extensions.other_extensions:
                  with push_extension(buf, extension_type):
                      buf.push_bytes(extension_value)
  
  
  CertificateEntry = Tuple[bytes, bytes]
  
  
  @dataclass
  class Certificate:
      request_context: bytes = b""
      certificates: List[CertificateEntry] = field(default_factory=list)
  
  
  def pull_certificate(buf: Buffer) -> Certificate:
      certificate = Certificate()
  
      assert buf.pull_uint8() == HandshakeType.CERTIFICATE
      with pull_block(buf, 3):
          certificate.request_context = pull_opaque(buf, 1)
  
          def pull_certificate_entry(buf: Buffer) -> CertificateEntry:
              data = pull_opaque(buf, 3)
              extensions = pull_opaque(buf, 2)
              return (data, extensions)
  
          certificate.certificates = pull_list(
              buf, 3, partial(pull_certificate_entry, buf)
          )
  
      return certificate
  
  
  def push_certificate(buf: Buffer, certificate: Certificate) -> None:
      buf.push_uint8(HandshakeType.CERTIFICATE)
      with push_block(buf, 3):
          push_opaque(buf, 1, certificate.request_context)
  
          def push_certificate_entry(buf: Buffer, entry: CertificateEntry) -> None:
              push_opaque(buf, 3, entry[0])
              push_opaque(buf, 2, entry[1])
  
          push_list(
              buf, 3, partial(push_certificate_entry, buf), certificate.certificates
          )
  
  
  @dataclass
  class CertificateVerify:
      algorithm: int
      signature: bytes
  
  
  def pull_certificate_verify(buf: Buffer) -> CertificateVerify:
      assert buf.pull_uint8() == HandshakeType.CERTIFICATE_VERIFY
      with pull_block(buf, 3):
          algorithm = buf.pull_uint16()
          signature = pull_opaque(buf, 2)
  
      return CertificateVerify(algorithm=algorithm, signature=signature)
  
  
  def push_certificate_verify(buf: Buffer, verify: CertificateVerify) -> None:
      buf.push_uint8(HandshakeType.CERTIFICATE_VERIFY)
      with push_block(buf, 3):
          buf.push_uint16(verify.algorithm)
          push_opaque(buf, 2, verify.signature)
  
  
  @dataclass
  class Finished:
      verify_data: bytes = b""
  
  
  def pull_finished(buf: Buffer) -> Finished:
      finished = Finished()
  
      assert buf.pull_uint8() == HandshakeType.FINISHED
      finished.verify_data = pull_opaque(buf, 3)
  
      return finished
  
  
  def push_finished(buf: Buffer, finished: Finished) -> None:
      buf.push_uint8(HandshakeType.FINISHED)
      push_opaque(buf, 3, finished.verify_data)
  
  
  # CONTEXT
  
  
  class KeySchedule:
      def __init__(self, cipher_suite: CipherSuite):
          self.algorithm = cipher_suite_hash(cipher_suite)
          self.cipher_suite = cipher_suite
          self.generation = 0
          self.hash = hashes.Hash(self.algorithm, default_backend())
          self.hash_empty_value = self.hash.copy().finalize()
          self.secret = bytes(self.algorithm.digest_size)
  
      def certificate_verify_data(self, context_string: bytes) -> bytes:
          return b" " * 64 + context_string + b"\x00" + self.hash.copy().finalize()
  
      def finished_verify_data(self, secret: bytes) -> bytes:
          hmac_key = hkdf_expand_label(
              algorithm=self.algorithm,
              secret=secret,
              label=b"finished",
              hash_value=b"",
              length=self.algorithm.digest_size,
          )
  
          h = hmac.HMAC(hmac_key, algorithm=self.algorithm, backend=default_backend())
          h.update(self.hash.copy().finalize())
          return h.finalize()
  
      def derive_secret(self, label: bytes) -> bytes:
          return hkdf_expand_label(
              algorithm=self.algorithm,
              secret=self.secret,
              label=label,
              hash_value=self.hash.copy().finalize(),
              length=self.algorithm.digest_size,
          )
  
      def extract(self, key_material: Optional[bytes] = None) -> None:
          if key_material is None:
              key_material = bytes(self.algorithm.digest_size)
  
          if self.generation:
              self.secret = hkdf_expand_label(
                  algorithm=self.algorithm,
                  secret=self.secret,
                  label=b"derived",
                  hash_value=self.hash_empty_value,
                  length=self.algorithm.digest_size,
              )
  
          self.generation += 1
          self.secret = hkdf_extract(
              algorithm=self.algorithm, salt=self.secret, key_material=key_material
          )
  
      def update_hash(self, data: bytes) -> None:
          self.hash.update(data)
  
  
  class KeyScheduleProxy:
      def __init__(self, cipher_suites: List[CipherSuite]):
          self.__schedules = dict(map(lambda c: (c, KeySchedule(c)), cipher_suites))
  
      def extract(self, key_material: Optional[bytes] = None) -> None:
          for k in self.__schedules.values():
              k.extract(key_material)
  
      def select(self, cipher_suite: CipherSuite) -> KeySchedule:
          return self.__schedules[cipher_suite]
  
      def update_hash(self, data: bytes) -> None:
          for k in self.__schedules.values():
              k.update_hash(data)
  
  
  CIPHER_SUITES = {
      CipherSuite.AES_128_GCM_SHA256: hashes.SHA256,
      CipherSuite.AES_256_GCM_SHA384: hashes.SHA384,
      CipherSuite.CHACHA20_POLY1305_SHA256: hashes.SHA256,
  }
  
  SIGNATURE_ALGORITHMS: Dict = {
      SignatureAlgorithm.ED25519: (None, hashes.SHA512),
      SignatureAlgorithm.ECDSA_SECP256R1_SHA256: (None, hashes.SHA256),
      SignatureAlgorithm.ECDSA_SECP384R1_SHA384: (None, hashes.SHA384),
      SignatureAlgorithm.ECDSA_SECP521R1_SHA512: (None, hashes.SHA512),
      SignatureAlgorithm.RSA_PKCS1_SHA1: (padding.PKCS1v15, hashes.SHA1),
      SignatureAlgorithm.RSA_PKCS1_SHA256: (padding.PKCS1v15, hashes.SHA256),
      SignatureAlgorithm.RSA_PKCS1_SHA384: (padding.PKCS1v15, hashes.SHA384),
      SignatureAlgorithm.RSA_PKCS1_SHA512: (padding.PKCS1v15, hashes.SHA512),
      SignatureAlgorithm.RSA_PSS_RSAE_SHA256: (padding.PSS, hashes.SHA256),
      SignatureAlgorithm.RSA_PSS_RSAE_SHA384: (padding.PSS, hashes.SHA384),
      SignatureAlgorithm.RSA_PSS_RSAE_SHA512: (padding.PSS, hashes.SHA512),
  }
  
  GROUP_TO_CURVE: Dict = {
      Group.SECP256R1: ec.SECP256R1,
      Group.SECP384R1: ec.SECP384R1,
      Group.SECP521R1: ec.SECP521R1,
  }
  CURVE_TO_GROUP = dict((v, k) for k, v in GROUP_TO_CURVE.items())
  
  
  def cipher_suite_hash(cipher_suite: CipherSuite) -> hashes.HashAlgorithm:
      return CIPHER_SUITES[cipher_suite]()
  
  
  def decode_public_key(
      key_share: KeyShareEntry,
  ) -> Union[ec.EllipticCurvePublicKey, x25519.X25519PublicKey, x448.X448PublicKey, None]:
      if key_share[0] == Group.X25519:
          return x25519.X25519PublicKey.from_public_bytes(key_share[1])
      elif key_share[0] == Group.X448:
          return x448.X448PublicKey.from_public_bytes(key_share[1])
      elif key_share[0] in GROUP_TO_CURVE:
          return ec.EllipticCurvePublicKey.from_encoded_point(
              GROUP_TO_CURVE[key_share[0]](), key_share[1]
          )
      else:
          return None
  
  
  def encode_public_key(
      public_key: Union[
          ec.EllipticCurvePublicKey, x25519.X25519PublicKey, x448.X448PublicKey
      ]
  ) -> KeyShareEntry:
      if isinstance(public_key, x25519.X25519PublicKey):
          return (Group.X25519, public_key.public_bytes(Encoding.Raw, PublicFormat.Raw))
      elif isinstance(public_key, x448.X448PublicKey):
          return (Group.X448, public_key.public_bytes(Encoding.Raw, PublicFormat.Raw))
      return (
          CURVE_TO_GROUP[public_key.curve.__class__],
          public_key.public_bytes(Encoding.X962, PublicFormat.UncompressedPoint),
      )
  
  
  def negotiate(
      supported: List[T], offered: Optional[List[Any]], exc: Optional[Alert] = None
  ) -> T:
      if offered is not None:
          for c in supported:
              if c in offered:
                  return c
  
      if exc is not None:
          raise exc
      return None
  
  
  def signature_algorithm_params(
      signature_algorithm: int,
  ) -> Union[Tuple[ec.ECDSA], Tuple[padding.AsymmetricPadding, hashes.HashAlgorithm]]:
      padding_cls, algorithm_cls = SIGNATURE_ALGORITHMS[signature_algorithm]
      algorithm = algorithm_cls()
      if padding_cls is None:
          return (ec.ECDSA(algorithm),)
      elif padding_cls == padding.PSS:
          padding_obj = padding_cls(
              mgf=padding.MGF1(algorithm), salt_length=algorithm.digest_size
          )
      else:
          padding_obj = padding_cls()
      return padding_obj, algorithm
  
  
  @contextmanager
  def push_message(
      key_schedule: Union[KeySchedule, KeyScheduleProxy], buf: Buffer
  ) -> Generator:
      hash_start = buf.tell()
      yield
      key_schedule.update_hash(buf.data_slice(hash_start, buf.tell()))
  
  
  # callback types
  
  
  @dataclass
  class SessionTicket:
      """
      A TLS session ticket for session resumption.
      """
  
      age_add: int
      cipher_suite: CipherSuite
      not_valid_after: datetime.datetime
      not_valid_before: datetime.datetime
      resumption_secret: bytes
      server_name: str
      ticket: bytes
  
      max_early_data_size: Optional[int] = None
      other_extensions: List[Tuple[int, bytes]] = field(default_factory=list)
  
      @property
      def is_valid(self) -> bool:
          now = utcnow()
          return now >= self.not_valid_before and now <= self.not_valid_after
  
      @property
      def obfuscated_age(self) -> int:
          age = int((utcnow() - self.not_valid_before).total_seconds() * 1000)
          return (age + self.age_add) % (1 << 32)
  
  
  AlpnHandler = Callable[[str], None]
  SessionTicketFetcher = Callable[[bytes], Optional[SessionTicket]]
  SessionTicketHandler = Callable[[SessionTicket], None]
  
  
  class Context:
      def __init__(
          self,
          is_client: bool,
          alpn_protocols: Optional[List[str]] = None,
          cadata: Optional[bytes] = None,
          cafile: Optional[str] = None,
          capath: Optional[str] = None,
          cipher_suites: Optional[List[CipherSuite]] = None,
          logger: Optional[Union[logging.Logger, logging.LoggerAdapter]] = None,
          max_early_data: Optional[int] = None,
          server_name: Optional[str] = None,
          verify_mode: Optional[int] = None,
      ):
          # configuration
          self._alpn_protocols = alpn_protocols
          self._cadata = cadata
          self._cafile = cafile
          self._capath = capath
          self.certificate: Optional[x509.Certificate] = None
          self.certificate_chain: List[x509.Certificate] = []
          self.certificate_private_key: Optional[
              Union[dsa.DSAPrivateKey, ec.EllipticCurvePrivateKey, rsa.RSAPrivateKey]
          ] = None
          self.handshake_extensions: List[Extension] = []
          self._max_early_data = max_early_data
          self.session_ticket: Optional[SessionTicket] = None
          self._server_name = server_name
          if verify_mode is not None:
              self._verify_mode = verify_mode
          else:
              self._verify_mode = ssl.CERT_REQUIRED if is_client else ssl.CERT_NONE
  
          # callbacks
          self.alpn_cb: Optional[AlpnHandler] = None
          self.get_session_ticket_cb: Optional[SessionTicketFetcher] = None
          self.new_session_ticket_cb: Optional[SessionTicketHandler] = None
          self.update_traffic_key_cb: Callable[
              [Direction, Epoch, CipherSuite, bytes], None
          ] = lambda d, e, c, s: None
  
          # supported parameters
          if cipher_suites is not None:
              self._cipher_suites = cipher_suites
          else:
              self._cipher_suites = [
                  CipherSuite.AES_256_GCM_SHA384,
                  CipherSuite.AES_128_GCM_SHA256,
                  CipherSuite.CHACHA20_POLY1305_SHA256,
              ]
          self._legacy_compression_methods: List[int] = [CompressionMethod.NULL]
          self._psk_key_exchange_modes: List[int] = [PskKeyExchangeMode.PSK_DHE_KE]
          self._signature_algorithms: List[int] = [
              SignatureAlgorithm.ED25519,
              SignatureAlgorithm.RSA_PSS_RSAE_SHA256,
              SignatureAlgorithm.ECDSA_SECP256R1_SHA256,
              SignatureAlgorithm.RSA_PKCS1_SHA256,
              SignatureAlgorithm.RSA_PKCS1_SHA1,
          ]
          self._supported_groups = []
          if default_backend().x25519_supported():
              self._supported_groups.append(Group.X25519)
          if default_backend().x448_supported():
              self._supported_groups.append(Group.X448)
          self._supported_groups.append(Group.SECP256R1)
          self._supported_versions = [TLS_VERSION_1_3]
  
          # state
          self.alpn_negotiated: Optional[str] = None
          self.early_data_accepted = False
          self.key_schedule: Optional[KeySchedule] = None
          self.received_extensions: Optional[List[Extension]] = None
          self._key_schedule_psk: Optional[KeySchedule] = None
          self._key_schedule_proxy: Optional[KeyScheduleProxy] = None
          self._new_session_ticket: Optional[NewSessionTicket] = None
          self._peer_certificate: Optional[x509.Certificate] = None
          self._peer_certificate_chain: List[x509.Certificate] = []
          self._receive_buffer = b""
          self._session_resumed = False
          self._enc_key: Optional[bytes] = None
          self._dec_key: Optional[bytes] = None
          self.__logger = logger
  
          self._ec_private_key: Optional[ec.EllipticCurvePrivateKey] = None
          self._x25519_private_key: Optional[x25519.X25519PrivateKey] = None
          self._x448_private_key: Optional[x448.X448PrivateKey] = None
  
          if is_client:
              self.client_random = os.urandom(32)
              self.legacy_session_id = b""
              self.state = State.CLIENT_HANDSHAKE_START
          else:
              self.client_random = None
              self.legacy_session_id = None
              self.state = State.SERVER_EXPECT_CLIENT_HELLO
  
      @property
      def session_resumed(self) -> bool:
          """
          Returns True if session resumption was successfully used.
          """
          return self._session_resumed
  
      def handle_message(
          self, input_data: bytes, output_buf: Dict[Epoch, Buffer]
      ) -> None:
          if self.state == State.CLIENT_HANDSHAKE_START:
              self._client_send_hello(output_buf[Epoch.INITIAL])
              return
  
          self._receive_buffer += input_data
          while len(self._receive_buffer) >= 4:
              # determine message length
              message_type = self._receive_buffer[0]
              message_length = 0
              for b in self._receive_buffer[1:4]:
                  message_length = (message_length << 8) | b
              message_length += 4
  
              # check message is complete
              if len(self._receive_buffer) < message_length:
                  break
              message = self._receive_buffer[:message_length]
              self._receive_buffer = self._receive_buffer[message_length:]
  
              input_buf = Buffer(data=message)
  
              # client states
  
              if self.state == State.CLIENT_EXPECT_SERVER_HELLO:
                  if message_type == HandshakeType.SERVER_HELLO:
                      self._client_handle_hello(input_buf, output_buf[Epoch.INITIAL])
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.CLIENT_EXPECT_ENCRYPTED_EXTENSIONS:
                  if message_type == HandshakeType.ENCRYPTED_EXTENSIONS:
                      self._client_handle_encrypted_extensions(input_buf)
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.CLIENT_EXPECT_CERTIFICATE_REQUEST_OR_CERTIFICATE:
                  if message_type == HandshakeType.CERTIFICATE:
                      self._client_handle_certificate(input_buf)
                  else:
                      # FIXME: handle certificate request
                      raise AlertUnexpectedMessage
              elif self.state == State.CLIENT_EXPECT_CERTIFICATE_VERIFY:
                  if message_type == HandshakeType.CERTIFICATE_VERIFY:
                      self._client_handle_certificate_verify(input_buf)
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.CLIENT_EXPECT_FINISHED:
                  if message_type == HandshakeType.FINISHED:
                      self._client_handle_finished(input_buf, output_buf[Epoch.HANDSHAKE])
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.CLIENT_POST_HANDSHAKE:
                  if message_type == HandshakeType.NEW_SESSION_TICKET:
                      self._client_handle_new_session_ticket(input_buf)
                  else:
                      raise AlertUnexpectedMessage
  
              # server states
  
              elif self.state == State.SERVER_EXPECT_CLIENT_HELLO:
                  if message_type == HandshakeType.CLIENT_HELLO:
                      self._server_handle_hello(
                          input_buf,
                          output_buf[Epoch.INITIAL],
                          output_buf[Epoch.HANDSHAKE],
                          output_buf[Epoch.ONE_RTT],
                      )
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.SERVER_EXPECT_FINISHED:
                  if message_type == HandshakeType.FINISHED:
                      self._server_handle_finished(input_buf, output_buf[Epoch.ONE_RTT])
                  else:
                      raise AlertUnexpectedMessage
              elif self.state == State.SERVER_POST_HANDSHAKE:
                  raise AlertUnexpectedMessage
  
              assert input_buf.eof()
  
      def _build_session_ticket(
          self, new_session_ticket: NewSessionTicket, other_extensions: List[Extension]
      ) -> SessionTicket:
          resumption_master_secret = self.key_schedule.derive_secret(b"res master")
          resumption_secret = hkdf_expand_label(
              algorithm=self.key_schedule.algorithm,
              secret=resumption_master_secret,
              label=b"resumption",
              hash_value=new_session_ticket.ticket_nonce,
              length=self.key_schedule.algorithm.digest_size,
          )
  
          timestamp = utcnow()
          return SessionTicket(
              age_add=new_session_ticket.ticket_age_add,
              cipher_suite=self.key_schedule.cipher_suite,
              max_early_data_size=new_session_ticket.max_early_data_size,
              not_valid_after=timestamp
              + datetime.timedelta(seconds=new_session_ticket.ticket_lifetime),
              not_valid_before=timestamp,
              other_extensions=other_extensions,
              resumption_secret=resumption_secret,
              server_name=self._server_name,
              ticket=new_session_ticket.ticket,
          )
  
      def _client_send_hello(self, output_buf: Buffer) -> None:
          key_share: List[KeyShareEntry] = []
          supported_groups: List[int] = []
  
          for group in self._supported_groups:
              if group == Group.SECP256R1:
                  self._ec_private_key = ec.generate_private_key(
                      GROUP_TO_CURVE[Group.SECP256R1](), default_backend()
                  )
                  key_share.append(encode_public_key(self._ec_private_key.public_key()))
                  supported_groups.append(Group.SECP256R1)
              elif group == Group.X25519:
                  self._x25519_private_key = x25519.X25519PrivateKey.generate()
                  key_share.append(
                      encode_public_key(self._x25519_private_key.public_key())
                  )
                  supported_groups.append(Group.X25519)
              elif group == Group.X448:
                  self._x448_private_key = x448.X448PrivateKey.generate()
                  key_share.append(encode_public_key(self._x448_private_key.public_key()))
                  supported_groups.append(Group.X448)
              elif group == Group.GREASE:
                  key_share.append((Group.GREASE, b"\x00"))
                  supported_groups.append(Group.GREASE)
  
          assert len(key_share), "no key share entries"
  
          hello = ClientHello(
              random=self.client_random,
              legacy_session_id=self.legacy_session_id,
              cipher_suites=[int(x) for x in self._cipher_suites],
              legacy_compression_methods=self._legacy_compression_methods,
              alpn_protocols=self._alpn_protocols,
              key_share=key_share,
              psk_key_exchange_modes=self._psk_key_exchange_modes
              if (self.session_ticket or self.new_session_ticket_cb is not None)
              else None,
              server_name=self._server_name,
              signature_algorithms=self._signature_algorithms,
              supported_groups=supported_groups,
              supported_versions=self._supported_versions,
              other_extensions=self.handshake_extensions,
          )
  
          # PSK
          if self.session_ticket and self.session_ticket.is_valid:
              self._key_schedule_psk = KeySchedule(self.session_ticket.cipher_suite)
              self._key_schedule_psk.extract(self.session_ticket.resumption_secret)
              binder_key = self._key_schedule_psk.derive_secret(b"res binder")
              binder_length = self._key_schedule_psk.algorithm.digest_size
  
              # update hello
              if self.session_ticket.max_early_data_size is not None:
                  hello.early_data = True
              hello.pre_shared_key = OfferedPsks(
                  identities=[
                      (self.session_ticket.ticket, self.session_ticket.obfuscated_age)
                  ],
                  binders=[bytes(binder_length)],
              )
  
              # serialize hello without binder
              tmp_buf = Buffer(capacity=1024)
              push_client_hello(tmp_buf, hello)
  
              # calculate binder
              hash_offset = tmp_buf.tell() - binder_length - 3
              self._key_schedule_psk.update_hash(tmp_buf.data_slice(0, hash_offset))
              binder = self._key_schedule_psk.finished_verify_data(binder_key)
              hello.pre_shared_key.binders[0] = binder
              self._key_schedule_psk.update_hash(
                  tmp_buf.data_slice(hash_offset, hash_offset + 3) + binder
              )
  
              # calculate early data key
              if hello.early_data:
                  early_key = self._key_schedule_psk.derive_secret(b"c e traffic")
                  self.update_traffic_key_cb(
                      Direction.ENCRYPT,
                      Epoch.ZERO_RTT,
                      self._key_schedule_psk.cipher_suite,
                      early_key,
                  )
  
          self._key_schedule_proxy = KeyScheduleProxy(self._cipher_suites)
          self._key_schedule_proxy.extract(None)
  
          with push_message(self._key_schedule_proxy, output_buf):
              push_client_hello(output_buf, hello)
  
          self._set_state(State.CLIENT_EXPECT_SERVER_HELLO)
  
      def _client_handle_hello(self, input_buf: Buffer, output_buf: Buffer) -> None:
          peer_hello = pull_server_hello(input_buf)
  
          cipher_suite = negotiate(
              self._cipher_suites,
              [peer_hello.cipher_suite],
              AlertHandshakeFailure("Unsupported cipher suite"),
          )
          assert peer_hello.compression_method in self._legacy_compression_methods
          assert peer_hello.supported_version in self._supported_versions
  
          # select key schedule
          if peer_hello.pre_shared_key is not None:
              if (
                  self._key_schedule_psk is None
                  or peer_hello.pre_shared_key != 0
                  or cipher_suite != self._key_schedule_psk.cipher_suite
              ):
                  raise AlertIllegalParameter
              self.key_schedule = self._key_schedule_psk
              self._session_resumed = True
          else:
              self.key_schedule = self._key_schedule_proxy.select(cipher_suite)
          self._key_schedule_psk = None
          self._key_schedule_proxy = None
  
          # perform key exchange
          peer_public_key = decode_public_key(peer_hello.key_share)
          shared_key: Optional[bytes] = None
          if (
              isinstance(peer_public_key, x25519.X25519PublicKey)
              and self._x25519_private_key is not None
          ):
              shared_key = self._x25519_private_key.exchange(peer_public_key)
          elif (
              isinstance(peer_public_key, x448.X448PublicKey)
              and self._x448_private_key is not None
          ):
              shared_key = self._x448_private_key.exchange(peer_public_key)
          elif (
              isinstance(peer_public_key, ec.EllipticCurvePublicKey)
              and self._ec_private_key is not None
              and self._ec_private_key.public_key().curve.__class__
              == peer_public_key.curve.__class__
          ):
              shared_key = self._ec_private_key.exchange(ec.ECDH(), peer_public_key)
          assert shared_key is not None
  
          self.key_schedule.update_hash(input_buf.data)
          self.key_schedule.extract(shared_key)
  
          self._setup_traffic_protection(
              Direction.DECRYPT, Epoch.HANDSHAKE, b"s hs traffic"
          )
  
          self._set_state(State.CLIENT_EXPECT_ENCRYPTED_EXTENSIONS)
  
      def _client_handle_encrypted_extensions(self, input_buf: Buffer) -> None:
          encrypted_extensions = pull_encrypted_extensions(input_buf)
  
          self.alpn_negotiated = encrypted_extensions.alpn_protocol
          self.early_data_accepted = encrypted_extensions.early_data
          self.received_extensions = encrypted_extensions.other_extensions
          if self.alpn_cb:
              self.alpn_cb(self.alpn_negotiated)
  
          self._setup_traffic_protection(
              Direction.ENCRYPT, Epoch.HANDSHAKE, b"c hs traffic"
          )
          self.key_schedule.update_hash(input_buf.data)
  
          # if the server accepted our PSK we are done, other we want its certificate
          if self._session_resumed:
              self._set_state(State.CLIENT_EXPECT_FINISHED)
          else:
              self._set_state(State.CLIENT_EXPECT_CERTIFICATE_REQUEST_OR_CERTIFICATE)
  
      def _client_handle_certificate(self, input_buf: Buffer) -> None:
          certificate = pull_certificate(input_buf)
  
          self._peer_certificate = x509.load_der_x509_certificate(
              certificate.certificates[0][0], backend=default_backend()
          )
          self._peer_certificate_chain = [
              x509.load_der_x509_certificate(
                  certificate.certificates[i][0], backend=default_backend()
              )
              for i in range(1, len(certificate.certificates))
          ]
  
          self.key_schedule.update_hash(input_buf.data)
  
          self._set_state(State.CLIENT_EXPECT_CERTIFICATE_VERIFY)
  
      def _client_handle_certificate_verify(self, input_buf: Buffer) -> None:
          verify = pull_certificate_verify(input_buf)
  
          assert verify.algorithm in self._signature_algorithms
  
          # check signature
          try:
              if isinstance(self._peer_certificate.public_key(), ed25519.Ed25519PublicKey):
                  self._peer_certificate.public_key().verify(
                      verify.signature,
                      self.key_schedule.certificate_verify_data(
                          b"TLS 1.3, server CertificateVerify"
                      ),
                  )
              else:
                  self._peer_certificate.public_key().verify(
                      verify.signature,
                      self.key_schedule.certificate_verify_data(
                          b"TLS 1.3, server CertificateVerify"
                      ),
                      *signature_algorithm_params(verify.algorithm),
                  )
          except InvalidSignature:
              raise AlertDecryptError
  
          # check certificate
          if self._verify_mode != ssl.CERT_NONE:
              verify_certificate(
                  cadata=self._cadata,
                  cafile=self._cafile,
                  capath=self._capath,
                  certificate=self._peer_certificate,
                  chain=self._peer_certificate_chain,
                  server_name=self._server_name,
              )
  
          self.key_schedule.update_hash(input_buf.data)
  
          self._set_state(State.CLIENT_EXPECT_FINISHED)
  
      def _client_handle_finished(self, input_buf: Buffer, output_buf: Buffer) -> None:
          finished = pull_finished(input_buf)
  
          # check verify data
          expected_verify_data = self.key_schedule.finished_verify_data(self._dec_key)
          if finished.verify_data != expected_verify_data:
              raise AlertDecryptError
          self.key_schedule.update_hash(input_buf.data)
  
          # prepare traffic keys
          assert self.key_schedule.generation == 2
          self.key_schedule.extract(None)
          self._setup_traffic_protection(
              Direction.DECRYPT, Epoch.ONE_RTT, b"s ap traffic"
          )
          next_enc_key = self.key_schedule.derive_secret(b"c ap traffic")
  
          # send finished
          with push_message(self.key_schedule, output_buf):
              push_finished(
                  output_buf,
                  Finished(
                      verify_data=self.key_schedule.finished_verify_data(self._enc_key)
                  ),
              )
  
          # commit traffic key
          self._enc_key = next_enc_key
          self.update_traffic_key_cb(
              Direction.ENCRYPT,
              Epoch.ONE_RTT,
              self.key_schedule.cipher_suite,
              self._enc_key,
          )
  
          self._set_state(State.CLIENT_POST_HANDSHAKE)
  
      def _client_handle_new_session_ticket(self, input_buf: Buffer) -> None:
          new_session_ticket = pull_new_session_ticket(input_buf)
  
          # notify application
          if self.new_session_ticket_cb is not None:
              ticket = self._build_session_ticket(
                  new_session_ticket, self.received_extensions
              )
              self.new_session_ticket_cb(ticket)
  
      def _server_handle_hello(
          self,
          input_buf: Buffer,
          initial_buf: Buffer,
          handshake_buf: Buffer,
          onertt_buf: Buffer,
      ) -> None:
          peer_hello = pull_client_hello(input_buf)
  
          # determine applicable signature algorithms
          signature_algorithms: List[SignatureAlgorithm] = []
          if isinstance(self.certificate_private_key, rsa.RSAPrivateKey):
              signature_algorithms = [
                  SignatureAlgorithm.RSA_PSS_RSAE_SHA256,
                  SignatureAlgorithm.RSA_PKCS1_SHA256,
                  SignatureAlgorithm.RSA_PKCS1_SHA1,
              ]
          elif isinstance(
              self.certificate_private_key, ec.EllipticCurvePrivateKey
          ) and isinstance(self.certificate_private_key.curve, ec.SECP256R1):
              signature_algorithms = [SignatureAlgorithm.ECDSA_SECP256R1_SHA256]
          elif isinstance(
              self.certificate_private_key, ed25519.Ed25519PrivateKey
          ):
              signature_algorithms = [SignatureAlgorithm.ED25519]
          # negotiate parameters
          cipher_suite = negotiate(
              self._cipher_suites,
              peer_hello.cipher_suites,
              AlertHandshakeFailure("No supported cipher suite"),
          )
          compression_method = negotiate(
              self._legacy_compression_methods,
              peer_hello.legacy_compression_methods,
              AlertHandshakeFailure("No supported compression method"),
          )
          psk_key_exchange_mode = negotiate(
              self._psk_key_exchange_modes, peer_hello.psk_key_exchange_modes
          )
  
          signature_algorithm = negotiate(
              signature_algorithms,
              peer_hello.signature_algorithms,
              AlertHandshakeFailure("No supported signature algorithm"),
          )
          supported_version = negotiate(
              self._supported_versions,
              peer_hello.supported_versions,
              AlertProtocolVersion("No supported protocol version"),
          )
  
          # negotiate ALPN
          if self._alpn_protocols is not None:
              self.alpn_negotiated = negotiate(
                  self._alpn_protocols,
                  peer_hello.alpn_protocols,
                  AlertHandshakeFailure("No common ALPN protocols"),
              )
          if self.alpn_cb:
              self.alpn_cb(self.alpn_negotiated)
  
          self.client_random = peer_hello.random
          self.server_random = os.urandom(32)
          self.legacy_session_id = peer_hello.legacy_session_id
          self.received_extensions = peer_hello.other_extensions
  
          # select key schedule
          pre_shared_key = None
          if (
              self.get_session_ticket_cb is not None
              and psk_key_exchange_mode is not None
              and peer_hello.pre_shared_key is not None
              and len(peer_hello.pre_shared_key.identities) == 1
              and len(peer_hello.pre_shared_key.binders) == 1
          ):
              # ask application to find session ticket
              identity = peer_hello.pre_shared_key.identities[0]
              session_ticket = self.get_session_ticket_cb(identity[0])
  
              # validate session ticket
              if (
                  session_ticket is not None
                  and session_ticket.is_valid
                  and session_ticket.cipher_suite == cipher_suite
              ):
                  self.key_schedule = KeySchedule(cipher_suite)
                  self.key_schedule.extract(session_ticket.resumption_secret)
  
                  binder_key = self.key_schedule.derive_secret(b"res binder")
                  binder_length = self.key_schedule.algorithm.digest_size
  
                  hash_offset = input_buf.tell() - binder_length - 3
                  binder = input_buf.data_slice(
                      hash_offset + 3, hash_offset + 3 + binder_length
                  )
  
                  self.key_schedule.update_hash(input_buf.data_slice(0, hash_offset))
                  expected_binder = self.key_schedule.finished_verify_data(binder_key)
  
                  if binder != expected_binder:
                      raise AlertHandshakeFailure("PSK validation failed")
  
                  self.key_schedule.update_hash(
                      input_buf.data_slice(hash_offset, hash_offset + 3 + binder_length)
                  )
                  self._session_resumed = True
  
                  # calculate early data key
                  if peer_hello.early_data:
                      early_key = self.key_schedule.derive_secret(b"c e traffic")
                      self.early_data_accepted = True
                      self.update_traffic_key_cb(
                          Direction.DECRYPT,
                          Epoch.ZERO_RTT,
                          self.key_schedule.cipher_suite,
                          early_key,
                      )
  
                  pre_shared_key = 0
  
          # if PSK is not used, initialize key schedule
          if pre_shared_key is None:
              self.key_schedule = KeySchedule(cipher_suite)
              self.key_schedule.extract(None)
              self.key_schedule.update_hash(input_buf.data)
  
          # perform key exchange
          public_key: Union[
              ec.EllipticCurvePublicKey, x25519.X25519PublicKey, x448.X448PublicKey
          ]
          shared_key: Optional[bytes] = None
          for key_share in peer_hello.key_share:
              peer_public_key = decode_public_key(key_share)
              if isinstance(peer_public_key, x25519.X25519PublicKey):
                  self._x25519_private_key = x25519.X25519PrivateKey.generate()
                  public_key = self._x25519_private_key.public_key()
                  shared_key = self._x25519_private_key.exchange(peer_public_key)
                  break
              elif isinstance(peer_public_key, x448.X448PublicKey):
                  self._x448_private_key = x448.X448PrivateKey.generate()
                  public_key = self._x448_private_key.public_key()
                  shared_key = self._x448_private_key.exchange(peer_public_key)
                  break
              elif isinstance(peer_public_key, ec.EllipticCurvePublicKey):
                  self._ec_private_key = ec.generate_private_key(
                      GROUP_TO_CURVE[key_share[0]](), default_backend()
                  )
                  public_key = self._ec_private_key.public_key()
                  shared_key = self._ec_private_key.exchange(ec.ECDH(), peer_public_key)
                  break
          assert shared_key is not None
  
          # send hello
          hello = ServerHello(
              random=self.server_random,
              legacy_session_id=self.legacy_session_id,
              cipher_suite=cipher_suite,
              compression_method=compression_method,
              key_share=encode_public_key(public_key),
              pre_shared_key=pre_shared_key,
              supported_version=supported_version,
          )
          with push_message(self.key_schedule, initial_buf):
              push_server_hello(initial_buf, hello)
          self.key_schedule.extract(shared_key)
  
          self._setup_traffic_protection(
              Direction.ENCRYPT, Epoch.HANDSHAKE, b"s hs traffic"
          )
          self._setup_traffic_protection(
              Direction.DECRYPT, Epoch.HANDSHAKE, b"c hs traffic"
          )
  
          # send encrypted extensions
          with push_message(self.key_schedule, handshake_buf):
              push_encrypted_extensions(
                  handshake_buf,
                  EncryptedExtensions(
                      alpn_protocol=self.alpn_negotiated,
                      early_data=self.early_data_accepted,
                      other_extensions=self.handshake_extensions,
                  ),
              )
  
          if pre_shared_key is None:
              # send certificate
              with push_message(self.key_schedule, handshake_buf):
                  push_certificate(
                      handshake_buf,
                      Certificate(
                          request_context=b"",
                          certificates=[
                              (x.public_bytes(Encoding.DER), b"")
                              for x in [self.certificate] + self.certificate_chain
                          ],
                      ),
                  )
  
              # send certificate verify
              if isinstance(self.certificate_private_key, ed25519.Ed25519PrivateKey):
                  signature = self.certificate_private_key.sign(
                      self.key_schedule.certificate_verify_data(
                          b"TLS 1.3, server CertificateVerify"
                      )
                  )
              else:
                  signature = self.certificate_private_key.sign(
                      self.key_schedule.certificate_verify_data(
                          b"TLS 1.3, server CertificateVerify"
                      ),
                      *signature_algorithm_params(signature_algorithm),
                  )
  
              with push_message(self.key_schedule, handshake_buf):
                  push_certificate_verify(
                      handshake_buf,
                      CertificateVerify(
                          algorithm=signature_algorithm, signature=signature
                      ),
                  )
  
          # send finished
          with push_message(self.key_schedule, handshake_buf):
              push_finished(
                  handshake_buf,
                  Finished(
                      verify_data=self.key_schedule.finished_verify_data(self._enc_key)
                  ),
              )
  
          # prepare traffic keys
          assert self.key_schedule.generation == 2
          self.key_schedule.extract(None)
          self._setup_traffic_protection(
              Direction.ENCRYPT, Epoch.ONE_RTT, b"s ap traffic"
          )
          self._next_dec_key = self.key_schedule.derive_secret(b"c ap traffic")
  
          # anticipate client's FINISHED as we don't use client auth
          self._expected_verify_data = self.key_schedule.finished_verify_data(
              self._dec_key
          )
          buf = Buffer(capacity=64)
          push_finished(buf, Finished(verify_data=self._expected_verify_data))
          self.key_schedule.update_hash(buf.data)
  
          # create a new session ticket
          if self.new_session_ticket_cb is not None and psk_key_exchange_mode is not None:
              self._new_session_ticket = NewSessionTicket(
                  ticket_lifetime=86400,
                  ticket_age_add=struct.unpack("I", os.urandom(4))[0],
                  ticket_nonce=b"",
                  ticket=os.urandom(64),
                  max_early_data_size=self._max_early_data,
              )
  
              # send messsage
              push_new_session_ticket(onertt_buf, self._new_session_ticket)
  
              # notify application
              ticket = self._build_session_ticket(
                  self._new_session_ticket, self.handshake_extensions
              )
              self.new_session_ticket_cb(ticket)
  
          self._set_state(State.SERVER_EXPECT_FINISHED)
  
      def _server_handle_finished(self, input_buf: Buffer, output_buf: Buffer) -> None:
          finished = pull_finished(input_buf)
  
          # check verify data
          if finished.verify_data != self._expected_verify_data:
              raise AlertDecryptError
  
          # commit traffic key
          self._dec_key = self._next_dec_key
          self._next_dec_key = None
          self.update_traffic_key_cb(
              Direction.DECRYPT,
              Epoch.ONE_RTT,
              self.key_schedule.cipher_suite,
              self._dec_key,
          )
  
          self._set_state(State.SERVER_POST_HANDSHAKE)
  
      def _setup_traffic_protection(
          self, direction: Direction, epoch: Epoch, label: bytes
      ) -> None:
          key = self.key_schedule.derive_secret(label)
  
          if direction == Direction.ENCRYPT:
              self._enc_key = key
          else:
              self._dec_key = key
  
          self.update_traffic_key_cb(
              direction, epoch, self.key_schedule.cipher_suite, key
          )
  
      def _set_state(self, state: State) -> None:
          if self.__logger:
              self.__logger.debug("TLS %s -> %s", self.state, state)
          self.state = state
  