def buffer_from_bytes(initializer):
    if sys.platform == 'win32':
        return ffi.new('unsigned char[]', initializer)
    return ffi.new('char[]', initializer) def buffer_from_unicode(initializer):
    return ffi.new('wchar_t []', initializer) def bytes_from_buffer(buffer, maxlen=None):
    if maxlen is not None:
        return ffi.buffer(buffer, maxlen)[:]
    return ffi.buffer(buffer)[:] def cast(library, type_, value):
    ffi_obj = _get_ffi(library)
    return ffi_obj.cast(type_, value) def deref(point):
    return point[0] def is_null(point):
    if point is None:
        return True
    if point == ffi.NULL:
        return True
    if ffi.getctype(ffi.typeof(point)) == 'void *':
        return False
    if point[0] == ffi.NULL:
        return True
    return False def native(type_, value):
    if type_ == str_cls:
        return ffi.string(value)
    if type_ == byte_cls:
        return ffi.buffer(value)[:]
    return type_(value) def new(library, type_, value=None):
    ffi_obj = _get_ffi(library)

    params = []
    if value is not None:
        params.append(value)
    if type_ in set(['BCRYPT_KEY_HANDLE', 'BCRYPT_ALG_HANDLE']):
        return ffi_obj.cast(type_, 0)
    return ffi_obj.new(type_, *params) def null():
    return ffi.NULL def ref(value, offset=0):
    return value + offset def sizeof(library, value):
    ffi_obj = _get_ffi(library)
    return ffi_obj.sizeof(value) def struct(library, name):
    ffi_obj = _get_ffi(library)
    return ffi_obj.new('%s *' % name) def unwrap(point):
    return point[0] def write_to_buffer(buffer, data, offset=0):
    buffer[offset:offset + len(data)] = data def handle_error(result, exception_class=None):
class Secur32Const():
    SCHANNEL_CRED_VERSION = 4
    SECPKG_CRED_OUTBOUND = 0x00000002
    UNISP_NAME = "Microsoft Unified Security Protocol Provider"
    SCH_CRED_MANUAL_CRED_VALIDATION = 0x00000008
    SCH_CRED_AUTO_CRED_VALIDATION = 0x00000020
    SCH_USE_STRONG_CRYPTO = 0x00400000
    SCH_CRED_NO_DEFAULT_CREDS = 0x00000010
    SECBUFFER_VERSION = 0
    SEC_E_OK = 0x00000000
    SEC_I_CONTINUE_NEEDED = 0x00090312
    SEC_I_CONTEXT_EXPIRED = 0x00090317
    SEC_I_RENEGOTIATE = 0x00090321
    SEC_E_INCOMPLETE_MESSAGE = 0x80090318
    SEC_E_INVALID_TOKEN = 0x80090308
    SEC_E_OUT_OF_SEQUENCE = 0x8009031
    SEC_E_MESSAGE_ALTERED = 0x8009030F
    SEC_E_CONTEXT_EXPIRED = 0x80090317
    SEC_E_INVALID_PARAMETER = 0x8009035D
    SEC_E_WRONG_PRINCIPAL = 0x80090322  # Domain name mismatch
    SEC_E_UNTRUSTED_ROOT = 0x80090325
    SEC_E_CERT_EXPIRED = 0x80090328
    SEC_E_ILLEGAL_MESSAGE = 0x80090326  # Handshake error
    SEC_E_INTERNAL_ERROR = 0x80090304  # Occurs when DH params are too small
    SEC_E_BUFFER_TOO_SMALL = 0x80090321
    SEC_I_INCOMPLETE_CREDENTIALS = 0x00090320
    ISC_REQ_REPLAY_DETECT = 4
    ISC_REQ_SEQUENCE_DETECT = 8
    ISC_REQ_CONFIDENTIALITY = 16
    ISC_REQ_ALLOCATE_MEMORY = 256
    ISC_REQ_INTEGRITY = 65536
    ISC_REQ_STREAM = 0x00008000
    ISC_REQ_USE_SUPPLIED_CREDS = 0x00000080
    ISC_RET_REPLAY_DETECT = 4
    ISC_RET_SEQUENCE_DETECT = 8
    ISC_RET_CONFIDENTIALITY = 16
    ISC_RET_ALLOCATED_MEMORY = 256
    ISC_RET_INTEGRITY = 65536
    ISC_RET_STREAM = 0x00008000
    SECBUFFER_ALERT = 17
    SECBUFFER_STREAM_HEADER = 7
    SECBUFFER_STREAM_TRAILER = 6
    SECBUFFER_EXTRA = 5
    SECBUFFER_TOKEN = 2
    SECBUFFER_DATA = 1
    SECBUFFER_EMPTY = 0
    SECPKG_ATTR_STREAM_SIZES = 0x04
    SECPKG_ATTR_CONNECTION_INFO = 0x5A
    SECPKG_ATTR_REMOTE_CERT_CONTEXT = 0x53
    SP_PROT_TLS1_2_CLIENT = 0x800
    SP_PROT_TLS1_1_CLIENT = 0x200
    SP_PROT_TLS1_CLIENT = 0x80
    SP_PROT_SSL3_CLIENT = 0x20
    SP_PROT_SSL2_CLIENT = 0x8
    CALG_AES_256 = 0x00006610
    CALG_AES_128 = 0x0000660E
    CALG_3DES = 0x00006603
    CALG_RC4 = 0x00006801
    CALG_RC2 = 0x00006602
    CALG_DES = 0x00006601
    CALG_MD5 = 0x00008003
    CALG_SHA1 = 0x00008004
    CALG_SHA256 = 0x0000800C
    CALG_SHA384 = 0x0000800D
    CALG_SHA512 = 0x0000800E
    CALG_DH_SF = 0x0000AA01
    CALG_DH_EPHEM = 0x0000AA02
    CALG_ECDH = 0x0000AA05
    CALG_ECDHE = 0x0000AE06
    CALG_RSA_KEYX = 0x0000A400
    CALG_RSA_SIGN = 0x00002400
    CALG_ECDSA = 0x00002203
    CALG_DSS_SIGN = 0x00002200 def handle_error(result):
class Crypt32Const():
    X509_ASN_ENCODING = 1
    ERROR_INSUFFICIENT_BUFFER = 122
    CERT_FIND_PROP_ONLY_ENHKEY_USAGE_FLAG = 0x4
    CRYPT_E_NOT_FOUND = -2146885628
    CERT_STORE_PROV_MEMORY = b'Memory'
    CERT_STORE_CREATE_NEW_FLAG = 0x00002000
    CERT_STORE_ADD_USE_EXISTING = 2
    USAGE_MATCH_TYPE_OR = 1
    CERT_CHAIN_POLICY_SSL = 4
    AUTHTYPE_SERVER = 2
    CERT_CHAIN_POLICY_ALLOW_UNKNOWN_CA_FLAG = 0x00000010
    CERT_CHAIN_POLICY_IGNORE_ALL_REV_UNKNOWN_FLAGS = 0x00000F00
    CERT_CHAIN_CACHE_END_CERT = 1
    CERT_CHAIN_REVOCATION_CHECK_CACHE_ONLY = 0x80000000
    TRUST_E_CERT_SIGNATURE = 0x80096004
    CERT_E_EXPIRED = 0x800B0101
    CERT_E_ROLE = 0x800B0103
    CERT_E_PURPOSE = 0x800B0106
    CERT_E_UNTRUSTEDROOT = 0x800B0109
    CERT_E_CN_NO_MATCH = 0x800B010F
    CRYPT_E_REVOKED = 0x80092010
    PKIX_KP_SERVER_AUTH = buffer_from_bytes(b"1.3.6.1.5.5.7.3.1\x00")
    SERVER_GATED_CRYPTO = buffer_from_bytes(b"1.3.6.1.4.1.311.10.3.3\x00")
    SGC_NETSCAPE = buffer_from_bytes(b"2.16.840.1.113730.4.1\x00") def handle_error(result): class TLSError(socket.error):

    """
    An exception related to TLS functionality
    """

    message = None

    def __init__(self, message):
        self.args = (message,)
        self.message = message

    def __str__(self):
        output = self.__unicode__()
        if sys.version_info < (3,):
            output = output.encode('utf-8')
        return output

    def __unicode__(self):
        return self.message class TLSVerificationError(TLSError):

    """
    A server certificate verification error happened during a TLS handshake
    """

    certificate = None

    def __init__(self, message, certificate):
        TLSError.__init__(self, message)
        self.certificate = certificate
        self.args = (message, certificate) class TLSDisconnectError(TLSConnectionError):
    pass class TLSGracefulDisconnectError(TLSDisconnectError):
    pass def detect_client_auth_request(server_handshake_bytes):
    """
    Determines if a CertificateRequest message is sent from the server asking
    the client for a certificate

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :return:
        A boolean - if a client certificate request was found
    """

    for record_type, _, record_data in parse_tls_records(server_handshake_bytes):
        if record_type != b'\x16':
            continue
        for message_type, message_data in parse_handshake_messages(record_data):
            if message_type == b'\x0d':
                return True
    return False def detect_other_protocol(server_handshake_bytes):
    """
    Looks at the server handshake bytes to try and detect a different protocol

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :return:
        None, or a unicode string of "ftp", "http", "imap", "pop3", "smtp"
    """

    if server_handshake_bytes[0:5] == b'HTTP/':
        return 'HTTP'

    if server_handshake_bytes[0:4] == b'220 ':
        if re.match(b'^[^\r\n]*ftp', server_handshake_bytes, re.I):
            return 'FTP'
        else:
            return 'SMTP'

    if server_handshake_bytes[0:4] == b'220-':
        return 'FTP'

    if server_handshake_bytes[0:4] == b'+OK ':
        return 'POP3'

    if server_handshake_bytes[0:4] == b'* OK' or server_handshake_bytes[0:9] == b'* PREAUTH':
        return 'IMAP'

    return None def extract_chain(server_handshake_bytes):
    """
    Extracts the X.509 certificates from the server handshake bytes for use
    when debugging

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :return:
        A list of asn1crypto.x509.Certificate objects
    """

    output = []

    chain_bytes = None

    for record_type, _, record_data in parse_tls_records(server_handshake_bytes):
        if record_type != b'\x16':
            continue
        for message_type, message_data in parse_handshake_messages(record_data):
            if message_type == b'\x0b':
                chain_bytes = message_data
                break
        if chain_bytes:
            break

    if chain_bytes:
        # The first 3 bytes are the cert chain length
        pointer = 3
        while pointer < len(chain_bytes):
            cert_length = int_from_bytes(chain_bytes[pointer:pointer + 3])
            cert_start = pointer + 3
            cert_end = cert_start + cert_length
            pointer = cert_end
            cert_bytes = chain_bytes[cert_start:cert_end]
            output.append(Certificate.load(cert_bytes))

    return output def get_dh_params_length(server_handshake_bytes):
    """
    Determines the length of the DH params from the ServerKeyExchange

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :return:
        None or an integer of the bit size of the DH parameters
    """

    output = None

    dh_params_bytes = None

    for record_type, _, record_data in parse_tls_records(server_handshake_bytes):
        if record_type != b'\x16':
            continue
        for message_type, message_data in parse_handshake_messages(record_data):
            if message_type == b'\x0c':
                dh_params_bytes = message_data
                break
        if dh_params_bytes:
            break

    if dh_params_bytes:
        output = int_from_bytes(dh_params_bytes[0:2]) * 8

    return output def parse_alert(server_handshake_bytes):
    """
    Parses the handshake for protocol alerts

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :return:
        None or an 2-element tuple of integers:
         0: 1 (warning) or 2 (fatal)
         1: The alert description (see https://tools.ietf.org/html/rfc5246#section-7.2)
    """

    for record_type, _, record_data in parse_tls_records(server_handshake_bytes):
        if record_type != b'\x15':
            continue
        if len(record_data) != 2:
            return None
        return (int_from_bytes(record_data[0:1]), int_from_bytes(record_data[1:2]))
    return None def parse_session_info(server_handshake_bytes, client_handshake_bytes):
    """
    Parse the TLS handshake from the client to the server to extract information
    including the cipher suite selected, if compression is enabled, the
    session id and if a new or reused session ticket exists.

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :param client_handshake_bytes:
        A byte string of the handshake data sent to the server

    :return:
        A dict with the following keys:
         - "protocol": unicode string
         - "cipher_suite": unicode string
         - "compression": boolean
         - "session_id": "new", "reused" or None
         - "session_ticket: "new", "reused" or None
    """

    protocol = None
    cipher_suite = None
    compression = False
    session_id = None
    session_ticket = None

    server_session_id = None
    client_session_id = None

    for record_type, _, record_data in parse_tls_records(server_handshake_bytes):
        if record_type != b'\x16':
            continue
        for message_type, message_data in parse_handshake_messages(record_data):
            # Ensure we are working with a ServerHello message
            if message_type != b'\x02':
                continue
            protocol = {
                b'\x03\x00': "SSLv3",
                b'\x03\x01': "TLSv1",
                b'\x03\x02': "TLSv1.1",
                b'\x03\x03': "TLSv1.2",
                b'\x03\x04': "TLSv1.3",
            }[message_data[0:2]]

            session_id_length = int_from_bytes(message_data[34:35])
            if session_id_length > 0:
                server_session_id = message_data[35:35 + session_id_length]

            cipher_suite_start = 35 + session_id_length
            cipher_suite_bytes = message_data[cipher_suite_start:cipher_suite_start + 2]
            cipher_suite = CIPHER_SUITE_MAP[cipher_suite_bytes]

            compression_start = cipher_suite_start + 2
            compression = message_data[compression_start:compression_start + 1] != b'\x00'

            extensions_length_start = compression_start + 1
            extensions_data = message_data[extensions_length_start:]
            for extension_type, extension_data in _parse_hello_extensions(extensions_data):
                if extension_type == 35:
                    session_ticket = "new"
                    break
            break

    for record_type, _, record_data in parse_tls_records(client_handshake_bytes):
        if record_type != b'\x16':
            continue
        for message_type, message_data in parse_handshake_messages(record_data):
            # Ensure we are working with a ClientHello message
            if message_type != b'\x01':
                continue

            session_id_length = int_from_bytes(message_data[34:35])
            if session_id_length > 0:
                client_session_id = message_data[35:35 + session_id_length]

            cipher_suite_start = 35 + session_id_length
            cipher_suite_length = int_from_bytes(message_data[cipher_suite_start:cipher_suite_start + 2])

            compression_start = cipher_suite_start + 2 + cipher_suite_length
            compression_length = int_from_bytes(message_data[compression_start:compression_start + 1])

            # On subsequent requests, the session ticket will only be seen
            # in the ClientHello message
            if server_session_id is None and session_ticket is None:
                extensions_length_start = compression_start + 1 + compression_length
                extensions_data = message_data[extensions_length_start:]
                for extension_type, extension_data in _parse_hello_extensions(extensions_data):
                    if extension_type == 35:
                        session_ticket = "reused"
                        break
            break

    if server_session_id is not None:
        if client_session_id is None:
            session_id = "new"
        else:
            if client_session_id != server_session_id:
                session_id = "new"
            else:
                session_id = "reused"

    return {
        "protocol": protocol,
        "cipher_suite": cipher_suite,
        "compression": compression,
        "session_id": session_id,
        "session_ticket": session_ticket,
    } def raise_client_auth():
    """
    Raises a TLSError indicating client authentication is required

    :raises:
        TLSError
    """

    message = 'TLS handshake failed - client authentication required'
    raise TLSError(message) def raise_dh_params():
    """
    Raises a TLSError due to weak DH params

    :raises:
        TLSError
    """

    raise TLSError('TLS handshake failed - weak DH parameters') def raise_disconnection():
    """
    Raises a TLSDisconnectError due to a disconnection

    :raises:
        TLSDisconnectError
    """

    raise TLSDisconnectError('The remote end closed the connection') def raise_expired_not_yet_valid(certificate):
    """
    Raises a TLSVerificationError due to certificate being expired, or not yet
    being valid

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    validity = certificate['tbs_certificate']['validity']
    not_after = validity['not_after'].native
    not_before = validity['not_before'].native

    now = datetime.now(timezone.utc)

    if not_before > now:
        formatted_before = not_before.strftime('%Y-%m-%d %H:%M:%SZ')
        message = 'Server certificate verification failed - certificate not valid until %s' % formatted_before
    elif not_after < now:
        formatted_after = not_after.strftime('%Y-%m-%d %H:%M:%SZ')
        message = 'Server certificate verification failed - certificate expired %s' % formatted_after

    raise TLSVerificationError(message, certificate) def raise_handshake():
    """
    Raises a TLSError due to a handshake error

    :raises:
        TLSError
    """

    raise TLSError('TLS handshake failed') def raise_hostname(certificate, hostname):
    """
    Raises a TLSVerificationError due to a hostname mismatch

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    is_ip = re.match('^\\d+\\.\\d+\\.\\d+\\.\\d+$', hostname) or hostname.find(':') != -1
    if is_ip:
        hostname_type = 'IP address %s' % hostname
    else:
        hostname_type = 'domain name %s' % hostname
    message = 'Server certificate verification failed - %s does not match' % hostname_type
    valid_ips = ', '.join(certificate.valid_ips)
    valid_domains = ', '.join(certificate.valid_domains)
    if valid_domains:
        message += ' valid domains: %s' % valid_domains
    if valid_domains and valid_ips:
        message += ' or'
    if valid_ips:
        message += ' valid IP addresses: %s' % valid_ips
    raise TLSVerificationError(message, certificate) def raise_no_issuer(certificate):
    """
    Raises a TLSVerificationError due to no issuer certificate found in trust
    roots

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    message = 'Server certificate verification failed - certificate issuer not found in trusted root certificate store'
    raise TLSVerificationError(message, certificate) def raise_protocol_error(server_handshake_bytes):
    """
    Raises a TLSError due to a protocol error

    :param server_handshake_bytes:
        A byte string of the handshake data received from the server

    :raises:
        TLSError
    """

    other_protocol = detect_other_protocol(server_handshake_bytes)

    if other_protocol:
        raise TLSError('TLS protocol error - server responded using %s' % other_protocol)

    raise TLSError('TLS protocol error - server responded using a different protocol') def raise_protocol_version():
    """
    Raises a TLSError due to a TLS version incompatibility

    :raises:
        TLSError
    """

    raise TLSError('TLS handshake failed - protocol version error') def raise_revoked(certificate):
    """
    Raises a TLSVerificationError due to the certificate being revoked

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    message = 'Server certificate verification failed - certificate has been revoked'
    raise TLSVerificationError(message, certificate) def raise_self_signed(certificate):
    """
    Raises a TLSVerificationError due to a self-signed certificate
    roots

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    message = 'Server certificate verification failed - certificate is self-signed'
    raise TLSVerificationError(message, certificate) def raise_verification(certificate):
    """
    Raises a generic TLSVerificationError

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    message = 'Server certificate verification failed'
    raise TLSVerificationError(message, certificate) def raise_weak_signature(certificate):
    """
    Raises a TLSVerificationError when a certificate uses a weak signature
    algorithm

    :param certificate:
        An asn1crypto.x509.Certificate object

    :raises:
        TLSVerificationError
    """

    message = 'Server certificate verification failed - weak certificate signature algorithm'
    raise TLSVerificationError(message, certificate) def load_certificate(source):
    """
    Loads an x509 certificate into a Certificate object

    :param source:
        A byte string of file contents or a unicode string filename

    :raises:
        ValueError - when any of the parameters contain an invalid value
        TypeError - when any of the parameters are of the wrong type
        OSError - when an error is returned by the OS crypto library

    :return:
        A Certificate object
    """

    if isinstance(source, Asn1Certificate):
        certificate = source

    elif isinstance(source, byte_cls):
        certificate = parse_certificate(source)

    elif isinstance(source, str_cls):
        with open(source, 'rb') as f:
            certificate = parse_certificate(f.read())

    else:
        raise TypeError(pretty_message(
            '''
            source must be a byte string, unicode string or
            asn1crypto.x509.Certificate object, not %s
            ''',
            type_name(source)
        ))

    return _load_key(certificate, Certificate) class Certificate(_WinKey, _CertificateBase):
    """
    Container for the OS crypto library representation of a certificate
    """

    _public_key = None
    _self_signed = None

    def __init__(self, key_handle, asn1):
        """
        :param key_handle:
            A CNG BCRYPT_KEY_HANDLE value (Vista and newer) or an HCRYPTKEY
            (XP and 2003) from loading/importing the certificate

        :param asn1:
            An asn1crypto.x509.Certificate object
        """

        _WinKey.__init__(self, key_handle, asn1)

    @property
    def public_key(self):
        """
        :return:
            The PublicKey object for the public key this certificate contains
        """

        if self._public_key is None:
            self._public_key = load_public_key(self.asn1['tbs_certificate']['subject_public_key_info'])
        return self._public_key

    @property
    def self_signed(self):
        """
        :return:
            A boolean - if the certificate is self-signed
        """

        if self._self_signed is None:
            self._self_signed = False
            if self.asn1.self_signed in set(['yes', 'maybe']):

                signature_algo = self.asn1['signature_algorithm'].signature_algo
                hash_algo = self.asn1['signature_algorithm'].hash_algo

                if signature_algo == 'rsassa_pkcs1v15':
                    verify_func = rsa_pkcs1v15_verify
                elif signature_algo == 'dsa':
                    verify_func = dsa_verify
                elif signature_algo == 'ecdsa':
                    verify_func = ecdsa_verify
                else:
                    raise OSError(pretty_message(
                        '''
                        Unable to verify the signature of the certificate since
                        it uses the unsupported algorithm %s
                        ''',
                        signature_algo
                    ))

                try:
                    verify_func(
                        self,
                        self.asn1['signature_value'].native,
                        self.asn1['tbs_certificate'].dump(),
                        hash_algo
                    )
                    self._self_signed = True
                except (SignatureError):
                    pass

        return self._self_signed
import sys
import re
import socket as socket_
import select
import numbers
from .._asn1 import Certificate as Asn1Certificate
from .._errors import pretty_message
from .._ffi import (
    buffer_from_bytes,
    buffer_from_unicode,
    bytes_from_buffer,
    cast,
    deref,
    is_null,
    native,
    new,
    null,
    ref,
    sizeof,
    struct,
    unwrap,
    write_to_buffer,
)
from ._secur32 import secur32, Secur32Const, handle_error
from ._crypt32 import crypt32, Crypt32Const, handle_error as handle_crypt32_error
from ._kernel32 import kernel32
from .._types import type_name, str_cls, byte_cls, int_types
from ..errors import TLSError, TLSVerificationError, TLSDisconnectError, TLSGracefulDisconnectError
from .._tls import (
    detect_client_auth_request,
    detect_other_protocol,
    extract_chain,
    get_dh_params_length,
    parse_alert,
    parse_session_info,
    raise_client_auth,
    raise_dh_params,
    raise_disconnection,
    raise_expired_not_yet_valid,
    raise_handshake,
    raise_hostname,
    raise_no_issuer,
    raise_protocol_error,
    raise_protocol_version,
    raise_revoked,
    raise_self_signed,
    raise_verification,
    raise_weak_signature,
)
from .asymmetric import load_certificate, Certificate
from ..keys import parse_certificate
        sec_buffer_desc_pointer = struct(secur32, 'SecBufferDesc')
        sec_buffer_desc = unwrap(sec_buffer_desc_pointer)

        sec_buffer_desc.ulVersion = Secur32Const.SECBUFFER_VERSION
        sec_buffer_desc.cBuffers = number
        sec_buffer_desc.pBuffers = buffers

        return (sec_buffer_desc_pointer, buffers)

    def _extra_trust_root_validation(self):
        """
        Manually invoked windows certificate chain builder and verification
        step when there are extra trust roots to include in the search process
        """

        store = None
        cert_chain_context_pointer = None

        try:
            # We set up an in-memory store to pass as an extra store to grab
            # certificates from when performing the verification
            store = crypt32.CertOpenStore(
                Crypt32Const.CERT_STORE_PROV_MEMORY,
                Crypt32Const.X509_ASN_ENCODING,
                null(),
                0,
                null()
            )
            if is_null(store):
                handle_crypt32_error(0)

            cert_hashes = set()
            for cert in self._session._extra_trust_roots:
                cert_data = cert.dump()
                result = crypt32.CertAddEncodedCertificateToStore(
                    store,
                    Crypt32Const.X509_ASN_ENCODING,
                    cert_data,
                    len(cert_data),
                    Crypt32Const.CERT_STORE_ADD_USE_EXISTING,
                    null()
                )
                if not result:
                    handle_crypt32_error(0)
                cert_hashes.add(cert.sha256)

            cert_context_pointer_pointer = new(crypt32, 'PCERT_CONTEXT *')
            result = secur32.QueryContextAttributesW(
                self._context_handle_pointer,
                Secur32Const.SECPKG_ATTR_REMOTE_CERT_CONTEXT,
                cert_context_pointer_pointer
            )
            handle_error(result)

            cert_context_pointer = unwrap(cert_context_pointer_pointer)
            cert_context_pointer = cast(crypt32, 'PCERT_CONTEXT', cert_context_pointer)

            # We have to do a funky shuffle here because FILETIME from kernel32
            # is different than FILETIME from crypt32 when using cffi. If we
            # overwrite the "now_pointer" variable, cffi releases the backing
            # memory and we end up getting a validation error about certificate
            # expiration time.
            orig_now_pointer = new(kernel32, 'FILETIME *')
            kernel32.GetSystemTimeAsFileTime(orig_now_pointer)
            now_pointer = cast(crypt32, 'FILETIME *', orig_now_pointer)

            usage_identifiers = new(crypt32, 'char *[3]')
            usage_identifiers[0] = cast(crypt32, 'char *', Crypt32Const.PKIX_KP_SERVER_AUTH)
            usage_identifiers[1] = cast(crypt32, 'char *', Crypt32Const.SERVER_GATED_CRYPTO)
            usage_identifiers[2] = cast(crypt32, 'char *', Crypt32Const.SGC_NETSCAPE)

            cert_enhkey_usage_pointer = struct(crypt32, 'CERT_ENHKEY_USAGE')
            cert_enhkey_usage = unwrap(cert_enhkey_usage_pointer)
            cert_enhkey_usage.cUsageIdentifier = 3
            cert_enhkey_usage.rgpszUsageIdentifier = cast(crypt32, 'char **', usage_identifiers)

            cert_usage_match_pointer = struct(crypt32, 'CERT_USAGE_MATCH')
            cert_usage_match = unwrap(cert_usage_match_pointer)
            cert_usage_match.dwType = Crypt32Const.USAGE_MATCH_TYPE_OR
            cert_usage_match.Usage = cert_enhkey_usage

            cert_chain_para_pointer = struct(crypt32, 'CERT_CHAIN_PARA')
            cert_chain_para = unwrap(cert_chain_para_pointer)
            cert_chain_para.RequestedUsage = cert_usage_match
            cert_chain_para_size = sizeof(crypt32, cert_chain_para)
            cert_chain_para.cbSize = cert_chain_para_size

            cert_chain_context_pointer_pointer = new(crypt32, 'PCERT_CHAIN_CONTEXT *')
            result = crypt32.CertGetCertificateChain(
                null(),
                cert_context_pointer,
                now_pointer,
                store,
                cert_chain_para_pointer,
                Crypt32Const.CERT_CHAIN_CACHE_END_CERT | Crypt32Const.CERT_CHAIN_REVOCATION_CHECK_CACHE_ONLY,
                null(),
                cert_chain_context_pointer_pointer
            )
            handle_crypt32_error(result)

            cert_chain_policy_para_flags = Crypt32Const.CERT_CHAIN_POLICY_IGNORE_ALL_REV_UNKNOWN_FLAGS

            cert_chain_context_pointer = unwrap(cert_chain_context_pointer_pointer)

            # Unwrap the chain and if the final element in the chain is one of
            # extra trust roots, set flags so that we trust the certificate even
            # though it is not in the Trusted Roots store
            cert_chain_context = unwrap(cert_chain_context_pointer)
            num_chains = native(int, cert_chain_context.cChain)
            if num_chains == 1:
                first_simple_chain_pointer = unwrap(cert_chain_context.rgpChain)
                first_simple_chain = unwrap(first_simple_chain_pointer)
                num_elements = native(int, first_simple_chain.cElement)
                last_element_pointer = first_simple_chain.rgpElement[num_elements - 1]
                last_element = unwrap(last_element_pointer)
                last_element_cert = unwrap(last_element.pCertContext)
                last_element_cert_data = bytes_from_buffer(
                    last_element_cert.pbCertEncoded,
                    native(int, last_element_cert.cbCertEncoded)
                )

                last_cert = Asn1Certificate.load(last_element_cert_data)