signature:
def _bcrypt_load_key ( key_object , key_info , container , curve_name ) :
    """Loads a certificate , public key or private key into a Certificate ,
 PublicKey or PrivateKey object via CNG
 : param key _ object :
 An asn1crypto . x509 . Certificate , asn1crypto . keys . PublicKeyInfo or
 asn1crypto . keys . PrivateKeyInfo object
 : param key _ info :
 An asn1crypto . keys . PublicKeyInfo or asn1crypto . keys . PrivateKeyInfo
 object
 : param container :
 The class of the object to hold the key _ handle
 : param curve _ name :
 None or a unicode string of the curve name for an EC key
 : raises :
 ValueError - when any of the parameters contain an invalid value
 TypeError - when any of the parameters are of the wrong type
 oscrypto . errors . AsymmetricKeyError - when the key is incompatible with the OS crypto library
 OSError - when an error is returned by the OS crypto library
 : return :
 A PrivateKey , PublicKey or Certificate object , based on container"""

implementation:
alg_handle = None
key_handle = None
key_type = 'public' if isinstance ( key_info , keys . PublicKeyInfo ) else 'private'
algo = key_info . algorithm
try :
    alg_selector = key_info . curve [ 1 ] if algo == 'ec' else algo
    alg_constant = { 'rsa' : BcryptConst . BCRYPT_RSA_ALGORITHM , 'dsa' : BcryptConst . BCRYPT_DSA_ALGORITHM , 'secp256r1' : BcryptConst . BCRYPT_ECDSA_P256_ALGORITHM , 'secp384r1' : BcryptConst . BCRYPT_ECDSA_P384_ALGORITHM , 'secp521r1' : BcryptConst . BCRYPT_ECDSA_P521_ALGORITHM , } [ alg_selector ]
    alg_handle = open_alg_handle ( alg_constant )
    if algo == 'rsa' :
        if key_type == 'public' :
            blob_type = BcryptConst . BCRYPT_RSAPUBLIC_BLOB
            magic = BcryptConst . BCRYPT_RSAPUBLIC_MAGIC
            parsed_key = key_info [ 'public_key' ] . parsed
            prime1_size = 0
            prime2_size = 0
        else :
            blob_type = BcryptConst . BCRYPT_RSAFULLPRIVATE_BLOB
            magic = BcryptConst . BCRYPT_RSAFULLPRIVATE_MAGIC
            parsed_key = key_info [ 'private_key' ] . parsed
            prime1 = int_to_bytes ( parsed_key [ 'prime1' ] . native )
            prime2 = int_to_bytes ( parsed_key [ 'prime2' ] . native )
            exponent1 = int_to_bytes ( parsed_key [ 'exponent1' ] . native )
            exponent2 = int_to_bytes ( parsed_key [ 'exponent2' ] . native )
            coefficient = int_to_bytes ( parsed_key [ 'coefficient' ] . native )
            private_exponent = int_to_bytes ( parsed_key [ 'private_exponent' ] . native )
            prime1_size = len ( prime1 )
            prime2_size = len ( prime2 )
        public_exponent = int_to_bytes ( parsed_key [ 'public_exponent' ] . native )
        modulus = int_to_bytes ( parsed_key [ 'modulus' ] . native )
        blob_struct_pointer = struct ( bcrypt , 'BCRYPT_RSAKEY_BLOB' )
        blob_struct = unwrap ( blob_struct_pointer )
        blob_struct . Magic = magic
        blob_struct . BitLength = key_info . bit_size
        blob_struct . cbPublicExp = len ( public_exponent )
        blob_struct . cbModulus = len ( modulus )
        blob_struct . cbPrime1 = prime1_size
        blob_struct . cbPrime2 = prime2_size
        blob = struct_bytes ( blob_struct_pointer ) + public_exponent + modulus
        if key_type == 'private' :
            blob += prime1 + prime2
            blob += fill_width ( exponent1 , prime1_size )
            blob += fill_width ( exponent2 , prime2_size )
            blob += fill_width ( coefficient , prime1_size )
            blob += fill_width ( private_exponent , len ( modulus ) )
    elif algo == 'dsa' :
        if key_type == 'public' :
            blob_type = BcryptConst . BCRYPT_DSA_PUBLIC_BLOB
            public_key = key_info [ 'public_key' ] . parsed . native
            params = key_info [ 'algorithm' ] [ 'parameters' ]
        else :
            blob_type = BcryptConst . BCRYPT_DSA_PRIVATE_BLOB
            public_key = key_info . public_key . native
            private_bytes = int_to_bytes ( key_info [ 'private_key' ] . parsed . native )
            params = key_info [ 'private_key_algorithm' ] [ 'parameters' ]
        public_bytes = int_to_bytes ( public_key )
        p = int_to_bytes ( params [ 'p' ] . native )
        g = int_to_bytes ( params [ 'g' ] . native )
        q = int_to_bytes ( params [ 'q' ] . native )
        if key_info . bit_size > 1024 :
            q_len = len ( q )
        else :
            q_len = 20
        key_width = max ( len ( public_bytes ) , len ( g ) , len ( p ) )
        public_bytes = fill_width ( public_bytes , key_width )
        p = fill_width ( p , key_width )
        g = fill_width ( g , key_width )
        q = fill_width ( q , q_len )
        # We don ' t know the count or seed , so we set them to the max value
        # since setting them to 0 results in a parameter error
        count = b'\xff' * 4
        seed = b'\xff' * q_len
        if key_info . bit_size > 1024 :
            if key_type == 'public' :
                magic = BcryptConst . BCRYPT_DSA_PUBLIC_MAGIC_V2
            else :
                magic = BcryptConst . BCRYPT_DSA_PRIVATE_MAGIC_V2
            blob_struct_pointer = struct ( bcrypt , 'BCRYPT_DSA_KEY_BLOB_V2' )
            blob_struct = unwrap ( blob_struct_pointer )
            blob_struct . dwMagic = magic
            blob_struct . cbKey = key_width
            # We don ' t know if SHA256 was used here , but the output is long
            # enough for the generation of q for the supported 2048/224,
            # 2048/256 and 3072/256 FIPS approved pairs
            blob_struct . hashAlgorithm = BcryptConst . DSA_HASH_ALGORITHM_SHA256
            blob_struct . standardVersion = BcryptConst . DSA_FIPS186_3
            blob_struct . cbSeedLength = q_len
            blob_struct . cbGroupSize = q_len
            blob_struct . Count = byte_array ( count )
            blob = struct_bytes ( blob_struct_pointer )
            blob += seed + q + p + g + public_bytes
            if key_type == 'private' :
                blob += fill_width ( private_bytes , q_len )
        else :
            if key_type == 'public' :
                magic = BcryptConst . BCRYPT_DSA_PUBLIC_MAGIC
            else :
                magic = BcryptConst . BCRYPT_DSA_PRIVATE_MAGIC
            blob_struct_pointer = struct ( bcrypt , 'BCRYPT_DSA_KEY_BLOB' )
            blob_struct = unwrap ( blob_struct_pointer )
            blob_struct . dwMagic = magic
            blob_struct . cbKey = key_width
            blob_struct . Count = byte_array ( count )
            blob_struct . Seed = byte_array ( seed )
            blob_struct . q = byte_array ( q )
            blob = struct_bytes ( blob_struct_pointer ) + p + g + public_bytes
            if key_type == 'private' :
                blob += fill_width ( private_bytes , q_len )
    elif algo == 'ec' :
        if key_type == 'public' :
            blob_type = BcryptConst . BCRYPT_ECCPUBLIC_BLOB
            public_key = key_info [ 'public_key' ]
        else :
            blob_type = BcryptConst . BCRYPT_ECCPRIVATE_BLOB
            public_key = key_info . public_key
            private_bytes = int_to_bytes ( key_info [ 'private_key' ] . parsed [ 'private_key' ] . native )
        blob_struct_pointer = struct ( bcrypt , 'BCRYPT_ECCKEY_BLOB' )
        blob_struct = unwrap ( blob_struct_pointer )
        magic = { ( 'public' , 'secp256r1' ) : BcryptConst . BCRYPT_ECDSA_PUBLIC_P256_MAGIC , ( 'public' , 'secp384r1' ) : BcryptConst . BCRYPT_ECDSA_PUBLIC_P384_MAGIC , ( 'public' , 'secp521r1' ) : BcryptConst . BCRYPT_ECDSA_PUBLIC_P521_MAGIC , ( 'private' , 'secp256r1' ) : BcryptConst . BCRYPT_ECDSA_PRIVATE_P256_MAGIC , ( 'private' , 'secp384r1' ) : BcryptConst . BCRYPT_ECDSA_PRIVATE_P384_MAGIC , ( 'private' , 'secp521r1' ) : BcryptConst . BCRYPT_ECDSA_PRIVATE_P521_MAGIC , } [ ( key_type , curve_name ) ]
        key_width = { 'secp256r1' : 32 , 'secp384r1' : 48 , 'secp521r1' : 66 } [ curve_name ]
        x , y = public_key . to_coords ( )
        x_bytes = int_to_bytes ( x )
        y_bytes = int_to_bytes ( y )
        x_bytes = fill_width ( x_bytes , key_width )
        y_bytes = fill_width ( y_bytes , key_width )
        blob_struct . dwMagic = magic
        blob_struct . cbKey = key_width
        blob = struct_bytes ( blob_struct_pointer ) + x_bytes + y_bytes
        if key_type == 'private' :
            blob += fill_width ( private_bytes , key_width )
    key_handle_pointer = new ( bcrypt , 'BCRYPT_KEY_HANDLE *' )
    res = bcrypt . BCryptImportKeyPair ( alg_handle , null ( ) , blob_type , key_handle_pointer , blob , len ( blob ) , BcryptConst . BCRYPT_NO_KEY_VALIDATION )
    handle_error ( res )
    key_handle = unwrap ( key_handle_pointer )
    return container ( key_handle , key_object )
finally :
    if alg_handle :
        close_alg_handle ( alg_handle )


signature:
def load_public_key ( source ) :
    """Loads a public key into a PublicKey object
 : param source :
 A byte string of file contents , a unicode string filename or an
 asn1crypto . keys . PublicKeyInfo object
 : raises :
 ValueError - when any of the parameters contain an invalid value
 TypeError - when any of the parameters are of the wrong type
 oscrypto . errors . AsymmetricKeyError - when the public key is incompatible with the OS crypto library
 OSError - when an error is returned by the OS crypto library
 : return :
 A PublicKey object"""

implementation:
if isinstance ( source , keys . PublicKeyInfo ) :
    public_key = source
elif isinstance ( source , byte_cls ) :
    public_key = parse_public ( source )
elif isinstance ( source , str_cls ) :
    with open ( source , 'rb' ) as f :
        public_key = parse_public ( f . read ( ) )
else :
    raise TypeError ( pretty_message ( '''
            source must be a byte string, unicode string or
            asn1crypto.keys.PublicKeyInfo object, not %s
            ''' , type_name ( public_key ) ) )
return _load_key ( public_key , PublicKey )


signature:
def load_pkcs12 ( source , password = None ) :
    """Loads a . p12 or . pfx file into a PrivateKey object and one or more
 Certificates objects
 : param source :
 A byte string of file contents or a unicode string filename
 : param password :
 A byte or unicode string to decrypt the PKCS12 file . Unicode strings
 will be encoded using UTF - 8.
 : raises :
 ValueError - when any of the parameters contain an invalid value
 TypeError - when any of the parameters are of the wrong type
 oscrypto . errors . AsymmetricKeyError - when a contained key is incompatible with the OS crypto library
 OSError - when an error is returned by the OS crypto library
 : return :
 A three - element tuple containing ( PrivateKey , Certificate , [ Certificate , . . . ] )"""

implementation:
if password is not None :
    if isinstance ( password , str_cls ) :
        password = password . encode ( 'utf-8' )
    if not isinstance ( password , byte_cls ) :
        raise TypeError ( pretty_message ( '''
                password must be a byte string, not %s
                ''' , type_name ( password ) ) )
if isinstance ( source , str_cls ) :
    with open ( source , 'rb' ) as f :
        source = f . read ( )
elif not isinstance ( source , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            source must be a byte string or a unicode string, not %s
            ''' , type_name ( source ) ) )
key_info , cert_info , extra_certs_info = parse_pkcs12 ( source , password )
key = None
cert = None
if key_info :
    key = _load_key ( key_info , PrivateKey )
if cert_info :
    cert = _load_key ( cert_info . public_key , Certificate )
extra_certs = [ _load_key ( info . public_key , Certificate ) for info in extra_certs_info ]
return ( key , cert , extra_certs )


signature:
def rsa_pkcs1v15_verify ( certificate_or_public_key , signature , data , hash_algorithm ) :
    """Verifies an RSASSA - PKCS - v1.5 signature .
 When the hash _ algorithm is " raw " , the operation is identical to RSA
 public key decryption . That is : the data is not hashed and no ASN . 1
 structure with an algorithm identifier of the hash algorithm is placed in
 the encrypted byte string .
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to verify the signature with
 : param signature :
 A byte string of the signature to verify
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : raises :
 oscrypto . errors . SignatureError - when the signature is determined to be invalid
 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"""

implementation:
if certificate_or_public_key . algorithm != 'rsa' :
    raise ValueError ( 'The key specified is not an RSA public key' )
return _verify ( certificate_or_public_key , signature , data , hash_algorithm )


signature:
def rsa_pss_verify ( certificate_or_public_key , signature , data , hash_algorithm ) :
    """Verifies an RSASSA - PSS signature . For the PSS padding the mask gen algorithm
 will be mgf1 using the same hash algorithm as the signature . The salt length
 with be the length of the hash algorithm , and the trailer field with be the
 standard 0xBC byte .
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to verify the signature with
 : param signature :
 A byte string of the signature to verify
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " or " sha512"
 : raises :
 oscrypto . errors . SignatureError - when the signature is determined to be invalid
 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"""

implementation:
if certificate_or_public_key . algorithm != 'rsa' :
    raise ValueError ( 'The key specified is not an RSA public key' )
return _verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding = True )


signature:
def _verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding = False ) :
    """Verifies an RSA , DSA or ECDSA signature
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to verify the signature with
 : param signature :
 A byte string of the signature to verify
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : raises :
 oscrypto . errors . SignatureError - when the signature is determined to be invalid
 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"""

implementation:
if not isinstance ( certificate_or_public_key , ( Certificate , PublicKey ) ) :
    raise TypeError ( pretty_message ( '''
            certificate_or_public_key must be an instance of the Certificate or
            PublicKey class, not %s
            ''' , type_name ( certificate_or_public_key ) ) )
if not isinstance ( signature , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            signature must be a byte string, not %s
            ''' , type_name ( signature ) ) )
if not isinstance ( data , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            data must be a byte string, not %s
            ''' , type_name ( data ) ) )
valid_hash_algorithms = set ( [ 'md5' , 'sha1' , 'sha256' , 'sha384' , 'sha512' ] )
if certificate_or_public_key . algorithm == 'rsa' and not rsa_pss_padding :
    valid_hash_algorithms |= set ( [ 'raw' ] )
if hash_algorithm not in valid_hash_algorithms :
    valid_hash_algorithms_error = '"md5", "sha1", "sha256", "sha384", "sha512"'
    if certificate_or_public_key . algorithm == 'rsa' and not rsa_pss_padding :
        valid_hash_algorithms_error += ', "raw"'
    raise ValueError ( pretty_message ( '''
            hash_algorithm must be one of %s, not %s
            ''' , valid_hash_algorithms_error , repr ( hash_algorithm ) ) )
if certificate_or_public_key . algorithm != 'rsa' and rsa_pss_padding is not False :
    raise ValueError ( pretty_message ( '''
            PSS padding may only be used with RSA keys - signing via a %s key
            was requested
            ''' , certificate_or_public_key . algorithm . upper ( ) ) )
if hash_algorithm == 'raw' :
    if len ( data ) > certificate_or_public_key . byte_size - 11 :
        raise ValueError ( pretty_message ( '''
                data must be 11 bytes shorter than the key size when
                hash_algorithm is "raw" - key size is %s bytes, but
                data is %s bytes long
                ''' , certificate_or_public_key . byte_size , len ( data ) ) )
if _backend == 'winlegacy' :
    if certificate_or_public_key . algorithm == 'ec' :
        return _pure_python_ecdsa_verify ( certificate_or_public_key , signature , data , hash_algorithm )
    return _advapi32_verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding )
return _bcrypt_verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding )


signature:
def _advapi32_verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding = False ) :
    """Verifies an RSA , DSA or ECDSA signature via CryptoAPI
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to verify the signature with
 : param signature :
 A byte string of the signature to verify
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : raises :
 oscrypto . errors . SignatureError - when the signature is determined to be invalid
 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"""

implementation:
algo = certificate_or_public_key . algorithm
if algo == 'rsa' and rsa_pss_padding :
    hash_length = { 'sha1' : 20 , 'sha224' : 28 , 'sha256' : 32 , 'sha384' : 48 , 'sha512' : 64 } . get ( hash_algorithm , 0 )
    decrypted_signature = raw_rsa_public_crypt ( certificate_or_public_key , signature )
    key_size = certificate_or_public_key . bit_size
    if not verify_pss_padding ( hash_algorithm , hash_length , key_size , data , decrypted_signature ) :
        raise SignatureError ( 'Signature is invalid' )
    return
if algo == 'rsa' and hash_algorithm == 'raw' :
    padded_plaintext = raw_rsa_public_crypt ( certificate_or_public_key , signature )
    try :
        plaintext = remove_pkcs1v15_signature_padding ( certificate_or_public_key . byte_size , padded_plaintext )
        if not constant_compare ( plaintext , data ) :
            raise ValueError ( )
    except ( ValueError ) :
        raise SignatureError ( 'Signature is invalid' )
    return
hash_handle = None
try :
    alg_id = { 'md5' : Advapi32Const . CALG_MD5 , 'sha1' : Advapi32Const . CALG_SHA1 , 'sha256' : Advapi32Const . CALG_SHA_256 , 'sha384' : Advapi32Const . CALG_SHA_384 , 'sha512' : Advapi32Const . CALG_SHA_512 , } [ hash_algorithm ]
    hash_handle_pointer = new ( advapi32 , 'HCRYPTHASH *' )
    res = advapi32 . CryptCreateHash ( certificate_or_public_key . context_handle , alg_id , null ( ) , 0 , hash_handle_pointer )
    handle_error ( res )
    hash_handle = unwrap ( hash_handle_pointer )
    res = advapi32 . CryptHashData ( hash_handle , data , len ( data ) , 0 )
    handle_error ( res )
    if algo == 'dsa' : # Windows doesn ' t use the ASN . 1 Sequence for DSA signatures ,
    # so we have to convert it here for the verification to work
        try :
            signature = algos . DSASignature . load ( signature ) . to_p1363 ( )
            # Switch the two integers so that the reversal later will
            # result in the correct order
            half_len = len ( signature ) // 2
            signature = signature [ half_len : ] + signature [ : half_len ]
        except ( ValueError , OverflowError , TypeError ) :
            raise SignatureError ( 'Signature is invalid' )
    # The CryptoAPI expects signatures to be in little endian byte order ,
    # which is the opposite of other systems , so we must reverse it
    reversed_signature = signature [ : : - 1 ]
    res = advapi32 . CryptVerifySignatureW ( hash_handle , reversed_signature , len ( signature ) , certificate_or_public_key . key_handle , null ( ) , 0 )
    handle_error ( res )
finally :
    if hash_handle :
        advapi32 . CryptDestroyHash ( hash_handle )


signature:
def _bcrypt_verify ( certificate_or_public_key , signature , data , hash_algorithm , rsa_pss_padding = False ) :
    """Verifies an RSA , DSA or ECDSA signature via CNG
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to verify the signature with
 : param signature :
 A byte string of the signature to verify
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : raises :
 oscrypto . errors . SignatureError - when the signature is determined to be invalid
 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"""

implementation:
if hash_algorithm == 'raw' :
    digest = data
else :
    hash_constant = { 'md5' : BcryptConst . BCRYPT_MD5_ALGORITHM , 'sha1' : BcryptConst . BCRYPT_SHA1_ALGORITHM , 'sha256' : BcryptConst . BCRYPT_SHA256_ALGORITHM , 'sha384' : BcryptConst . BCRYPT_SHA384_ALGORITHM , 'sha512' : BcryptConst . BCRYPT_SHA512_ALGORITHM } [ hash_algorithm ]
    digest = getattr ( hashlib , hash_algorithm ) ( data ) . digest ( )
padding_info = null ( )
flags = 0
if certificate_or_public_key . algorithm == 'rsa' :
    if rsa_pss_padding :
        flags = BcryptConst . BCRYPT_PAD_PSS
        padding_info_struct_pointer = struct ( bcrypt , 'BCRYPT_PSS_PADDING_INFO' )
        padding_info_struct = unwrap ( padding_info_struct_pointer )
        # This has to be assigned to a variable to prevent cffi from gc ' ing it
        hash_buffer = buffer_from_unicode ( hash_constant )
        padding_info_struct . pszAlgId = cast ( bcrypt , 'wchar_t *' , hash_buffer )
        padding_info_struct . cbSalt = len ( digest )
    else :
        flags = BcryptConst . BCRYPT_PAD_PKCS1
        padding_info_struct_pointer = struct ( bcrypt , 'BCRYPT_PKCS1_PADDING_INFO' )
        padding_info_struct = unwrap ( padding_info_struct_pointer )
        # This has to be assigned to a variable to prevent cffi from gc ' ing it
        if hash_algorithm == 'raw' :
            padding_info_struct . pszAlgId = null ( )
        else :
            hash_buffer = buffer_from_unicode ( hash_constant )
            padding_info_struct . pszAlgId = cast ( bcrypt , 'wchar_t *' , hash_buffer )
    padding_info = cast ( bcrypt , 'void *' , padding_info_struct_pointer )
else : # Windows doesn ' t use the ASN . 1 Sequence for DSA / ECDSA signatures ,
# so we have to convert it here for the verification to work
    try :
        signature = algos . DSASignature . load ( signature ) . to_p1363 ( )
    except ( ValueError , OverflowError , TypeError ) :
        raise SignatureError ( 'Signature is invalid' )
res = bcrypt . BCryptVerifySignature ( certificate_or_public_key . key_handle , padding_info , digest , len ( digest ) , signature , len ( signature ) , flags )
failure = res == BcryptConst . STATUS_INVALID_SIGNATURE
failure = failure or res == BcryptConst . STATUS_INVALID_PARAMETER
if failure :
    raise SignatureError ( 'Signature is invalid' )
handle_error ( res )


signature:
def rsa_pkcs1v15_sign ( private_key , data , hash_algorithm ) :
    """Generates an RSASSA - PKCS - v1.5 signature .
 When the hash _ algorithm is " raw " , the operation is identical to RSA
 private key encryption . That is : the data is not hashed and no ASN . 1
 structure with an algorithm identifier of the hash algorithm is placed in
 the encrypted byte string .
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : 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 byte string of the signature"""

implementation:
if private_key . algorithm != 'rsa' :
    raise ValueError ( 'The key specified is not an RSA private key' )
return _sign ( private_key , data , hash_algorithm )


signature:
def rsa_pss_sign ( private_key , data , hash_algorithm ) :
    """Generates an RSASSA - PSS signature . For the PSS padding the mask gen
 algorithm will be mgf1 using the same hash algorithm as the signature . The
 salt length with be the length of the hash algorithm , and the trailer field
 with be the standard 0xBC byte .
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " or " sha512"
 : 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 byte string of the signature"""

implementation:
if private_key . algorithm != 'rsa' :
    raise ValueError ( 'The key specified is not an RSA private key' )
return _sign ( private_key , data , hash_algorithm , rsa_pss_padding = True )


signature:
def dsa_sign ( private_key , data , hash_algorithm ) :
    """Generates a DSA signature
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " or " sha512"
 : 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 byte string of the signature"""

implementation:
if private_key . algorithm != 'dsa' :
    raise ValueError ( 'The key specified is not a DSA private key' )
return _sign ( private_key , data , hash_algorithm )


signature:
def ecdsa_sign ( private_key , data , hash_algorithm ) :
    """Generates an ECDSA signature
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " or " sha512"
 : 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 byte string of the signature"""

implementation:
if private_key . algorithm != 'ec' :
    raise ValueError ( 'The key specified is not an EC private key' )
return _sign ( private_key , data , hash_algorithm )


signature:
def _sign ( private_key , data , hash_algorithm , rsa_pss_padding = False ) :
    """Generates an RSA , DSA or ECDSA signature
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : 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 byte string of the signature"""

implementation:
if not isinstance ( private_key , PrivateKey ) :
    raise TypeError ( pretty_message ( '''
            private_key must be an instance of PrivateKey, not %s
            ''' , type_name ( private_key ) ) )
if not isinstance ( data , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            data must be a byte string, not %s
            ''' , type_name ( data ) ) )
valid_hash_algorithms = set ( [ 'md5' , 'sha1' , 'sha256' , 'sha384' , 'sha512' ] )
if private_key . algorithm == 'rsa' and not rsa_pss_padding :
    valid_hash_algorithms |= set ( [ 'raw' ] )
if hash_algorithm not in valid_hash_algorithms :
    valid_hash_algorithms_error = '"md5", "sha1", "sha256", "sha384", "sha512"'
    if private_key . algorithm == 'rsa' and not rsa_pss_padding :
        valid_hash_algorithms_error += ', "raw"'
    raise ValueError ( pretty_message ( '''
            hash_algorithm must be one of %s, not %s
            ''' , valid_hash_algorithms_error , repr ( hash_algorithm ) ) )
if private_key . algorithm != 'rsa' and rsa_pss_padding is not False :
    raise ValueError ( pretty_message ( '''
            PSS padding may only be used with RSA keys - signing via a %s key
            was requested
            ''' , private_key . algorithm . upper ( ) ) )
if hash_algorithm == 'raw' :
    if len ( data ) > private_key . byte_size - 11 :
        raise ValueError ( pretty_message ( '''
                data must be 11 bytes shorter than the key size when
                hash_algorithm is "raw" - key size is %s bytes, but data
                is %s bytes long
                ''' , private_key . byte_size , len ( data ) ) )
if _backend == 'winlegacy' :
    if private_key . algorithm == 'ec' :
        return _pure_python_ecdsa_sign ( private_key , data , hash_algorithm )
    return _advapi32_sign ( private_key , data , hash_algorithm , rsa_pss_padding )
return _bcrypt_sign ( private_key , data , hash_algorithm , rsa_pss_padding )


signature:
def _advapi32_sign ( private_key , data , hash_algorithm , rsa_pss_padding = False ) :
    """Generates an RSA , DSA or ECDSA signature via CryptoAPI
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : 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 byte string of the signature"""

implementation:
algo = private_key . algorithm
if algo == 'rsa' and hash_algorithm == 'raw' :
    padded_data = add_pkcs1v15_signature_padding ( private_key . byte_size , data )
    return raw_rsa_private_crypt ( private_key , padded_data )
if algo == 'rsa' and rsa_pss_padding :
    hash_length = { 'sha1' : 20 , 'sha224' : 28 , 'sha256' : 32 , 'sha384' : 48 , 'sha512' : 64 } . get ( hash_algorithm , 0 )
    padded_data = add_pss_padding ( hash_algorithm , hash_length , private_key . bit_size , data )
    return raw_rsa_private_crypt ( private_key , padded_data )
if private_key . algorithm == 'dsa' and hash_algorithm == 'md5' :
    raise ValueError ( pretty_message ( '''
            Windows does not support md5 signatures with DSA keys
            ''' ) )
hash_handle = None
try :
    alg_id = { 'md5' : Advapi32Const . CALG_MD5 , 'sha1' : Advapi32Const . CALG_SHA1 , 'sha256' : Advapi32Const . CALG_SHA_256 , 'sha384' : Advapi32Const . CALG_SHA_384 , 'sha512' : Advapi32Const . CALG_SHA_512 , } [ hash_algorithm ]
    hash_handle_pointer = new ( advapi32 , 'HCRYPTHASH *' )
    res = advapi32 . CryptCreateHash ( private_key . context_handle , alg_id , null ( ) , 0 , hash_handle_pointer )
    handle_error ( res )
    hash_handle = unwrap ( hash_handle_pointer )
    res = advapi32 . CryptHashData ( hash_handle , data , len ( data ) , 0 )
    handle_error ( res )
    out_len = new ( advapi32 , 'DWORD *' )
    res = advapi32 . CryptSignHashW ( hash_handle , Advapi32Const . AT_SIGNATURE , null ( ) , 0 , null ( ) , out_len )
    handle_error ( res )
    buffer_length = deref ( out_len )
    buffer_ = buffer_from_bytes ( buffer_length )
    res = advapi32 . CryptSignHashW ( hash_handle , Advapi32Const . AT_SIGNATURE , null ( ) , 0 , buffer_ , out_len )
    handle_error ( res )
    output = bytes_from_buffer ( buffer_ , deref ( out_len ) )
    # CryptoAPI outputs the signature in little endian byte order , so we
    # must swap it for compatibility with other systems
    output = output [ : : - 1 ]
    if algo == 'dsa' : # Switch the two integers because the reversal just before switched
    # then
        half_len = len ( output ) // 2
        output = output [ half_len : ] + output [ : half_len ]
        # Windows doesn ' t use the ASN . 1 Sequence for DSA signatures ,
        # so we have to convert it here for the verification to work
        output = algos . DSASignature . from_p1363 ( output ) . dump ( )
    return output
finally :
    if hash_handle :
        advapi32 . CryptDestroyHash ( hash_handle )


signature:
def _bcrypt_sign ( private_key , data , hash_algorithm , rsa_pss_padding = False ) :
    """Generates an RSA , DSA or ECDSA signature via CNG
 : param private _ key :
 The PrivateKey to generate the signature with
 : param data :
 A byte string of the data the signature is for
 : param hash _ algorithm :
 A unicode string of " md5 " , " sha1 " , " sha256 " , " sha384 " , " sha512 " or " raw "
 : param rsa _ pss _ padding :
 If PSS padding should be used for RSA keys
 : 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 byte string of the signature"""

implementation:
if hash_algorithm == 'raw' :
    digest = data
else :
    hash_constant = { 'md5' : BcryptConst . BCRYPT_MD5_ALGORITHM , 'sha1' : BcryptConst . BCRYPT_SHA1_ALGORITHM , 'sha256' : BcryptConst . BCRYPT_SHA256_ALGORITHM , 'sha384' : BcryptConst . BCRYPT_SHA384_ALGORITHM , 'sha512' : BcryptConst . BCRYPT_SHA512_ALGORITHM } [ hash_algorithm ]
    digest = getattr ( hashlib , hash_algorithm ) ( data ) . digest ( )
padding_info = null ( )
flags = 0
if private_key . algorithm == 'rsa' :
    if rsa_pss_padding :
        hash_length = { 'md5' : 16 , 'sha1' : 20 , 'sha256' : 32 , 'sha384' : 48 , 'sha512' : 64 } [ hash_algorithm ]
        flags = BcryptConst . BCRYPT_PAD_PSS
        padding_info_struct_pointer = struct ( bcrypt , 'BCRYPT_PSS_PADDING_INFO' )
        padding_info_struct = unwrap ( padding_info_struct_pointer )
        # This has to be assigned to a variable to prevent cffi from gc ' ing it
        hash_buffer = buffer_from_unicode ( hash_constant )
        padding_info_struct . pszAlgId = cast ( bcrypt , 'wchar_t *' , hash_buffer )
        padding_info_struct . cbSalt = hash_length
    else :
        flags = BcryptConst . BCRYPT_PAD_PKCS1
        padding_info_struct_pointer = struct ( bcrypt , 'BCRYPT_PKCS1_PADDING_INFO' )
        padding_info_struct = unwrap ( padding_info_struct_pointer )
        # This has to be assigned to a variable to prevent cffi from gc ' ing it
        if hash_algorithm == 'raw' :
            padding_info_struct . pszAlgId = null ( )
        else :
            hash_buffer = buffer_from_unicode ( hash_constant )
            padding_info_struct . pszAlgId = cast ( bcrypt , 'wchar_t *' , hash_buffer )
    padding_info = cast ( bcrypt , 'void *' , padding_info_struct_pointer )
if private_key . algorithm == 'dsa' and private_key . bit_size > 1024 and hash_algorithm in set ( [ 'md5' , 'sha1' ] ) :
    raise ValueError ( pretty_message ( '''
            Windows does not support sha1 signatures with DSA keys based on
            sha224, sha256 or sha512
            ''' ) )
out_len = new ( bcrypt , 'DWORD *' )
res = bcrypt . BCryptSignHash ( private_key . key_handle , padding_info , digest , len ( digest ) , null ( ) , 0 , out_len , flags )
handle_error ( res )
buffer_len = deref ( out_len )
buffer = buffer_from_bytes ( buffer_len )
if private_key . algorithm == 'rsa' :
    padding_info = cast ( bcrypt , 'void *' , padding_info_struct_pointer )
res = bcrypt . BCryptSignHash ( private_key . key_handle , padding_info , digest , len ( digest ) , buffer , buffer_len , out_len , flags )
handle_error ( res )
signature = bytes_from_buffer ( buffer , deref ( out_len ) )
if private_key . algorithm != 'rsa' : # Windows doesn ' t use the ASN . 1 Sequence for DSA / ECDSA signatures ,
# so we have to convert it here for the verification to work
    signature = algos . DSASignature . from_p1363 ( signature ) . dump ( )
return signature


signature:
def _encrypt ( certificate_or_public_key , data , rsa_oaep_padding = False ) :
    """Encrypts a value using an RSA public key
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to encrypt with
 : param data :
 A byte string of the data to encrypt
 : param rsa _ oaep _ padding :
 If OAEP padding should be used instead of PKCS # 1 v1.5
 : 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 byte string of the ciphertext"""

implementation:
if not isinstance ( certificate_or_public_key , ( Certificate , PublicKey ) ) :
    raise TypeError ( pretty_message ( '''
            certificate_or_public_key must be an instance of the Certificate or
            PublicKey class, not %s
            ''' , type_name ( certificate_or_public_key ) ) )
if not isinstance ( data , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            data must be a byte string, not %s
            ''' , type_name ( data ) ) )
if not isinstance ( rsa_oaep_padding , bool ) :
    raise TypeError ( pretty_message ( '''
            rsa_oaep_padding must be a bool, not %s
            ''' , type_name ( rsa_oaep_padding ) ) )
if _backend == 'winlegacy' :
    return _advapi32_encrypt ( certificate_or_public_key , data , rsa_oaep_padding )
return _bcrypt_encrypt ( certificate_or_public_key , data , rsa_oaep_padding )


signature:
def _advapi32_encrypt ( certificate_or_public_key , data , rsa_oaep_padding = False ) :
    """Encrypts a value using an RSA public key via CryptoAPI
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to encrypt with
 : param data :
 A byte string of the data to encrypt
 : param rsa _ oaep _ padding :
 If OAEP padding should be used instead of PKCS # 1 v1.5
 : 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 byte string of the ciphertext"""

implementation:
flags = 0
if rsa_oaep_padding :
    flags = Advapi32Const . CRYPT_OAEP
out_len = new ( advapi32 , 'DWORD *' , len ( data ) )
res = advapi32 . CryptEncrypt ( certificate_or_public_key . ex_key_handle , null ( ) , True , flags , null ( ) , out_len , 0 )
handle_error ( res )
buffer_len = deref ( out_len )
buffer = buffer_from_bytes ( buffer_len )
write_to_buffer ( buffer , data )
pointer_set ( out_len , len ( data ) )
res = advapi32 . CryptEncrypt ( certificate_or_public_key . ex_key_handle , null ( ) , True , flags , buffer , out_len , buffer_len )
handle_error ( res )
return bytes_from_buffer ( buffer , deref ( out_len ) ) [ : : - 1 ]


signature:
def _bcrypt_encrypt ( certificate_or_public_key , data , rsa_oaep_padding = False ) :
    """Encrypts a value using an RSA public key via CNG
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey instance to encrypt with
 : param data :
 A byte string of the data to encrypt
 : param rsa _ oaep _ padding :
 If OAEP padding should be used instead of PKCS # 1 v1.5
 : 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 byte string of the ciphertext"""

implementation:
flags = BcryptConst . BCRYPT_PAD_PKCS1
if rsa_oaep_padding is True :
    flags = BcryptConst . BCRYPT_PAD_OAEP
    padding_info_struct_pointer = struct ( bcrypt , 'BCRYPT_OAEP_PADDING_INFO' )
    padding_info_struct = unwrap ( padding_info_struct_pointer )
    # This has to be assigned to a variable to prevent cffi from gc ' ing it
    hash_buffer = buffer_from_unicode ( BcryptConst . BCRYPT_SHA1_ALGORITHM )
    padding_info_struct . pszAlgId = cast ( bcrypt , 'wchar_t *' , hash_buffer )
    padding_info_struct . pbLabel = null ( )
    padding_info_struct . cbLabel = 0
    padding_info = cast ( bcrypt , 'void *' , padding_info_struct_pointer )
else :
    padding_info = null ( )
out_len = new ( bcrypt , 'ULONG *' )
res = bcrypt . BCryptEncrypt ( certificate_or_public_key . key_handle , data , len ( data ) , padding_info , null ( ) , 0 , null ( ) , 0 , out_len , flags )
handle_error ( res )
buffer_len = deref ( out_len )
buffer = buffer_from_bytes ( buffer_len )
res = bcrypt . BCryptEncrypt ( certificate_or_public_key . key_handle , data , len ( data ) , padding_info , null ( ) , 0 , buffer , buffer_len , out_len , flags )
handle_error ( res )
return bytes_from_buffer ( buffer , deref ( out_len ) )


signature:
def _decrypt ( private_key , ciphertext , rsa_oaep_padding = False ) :
    """Encrypts a value using an RSA private key
 : param private _ key :
 A PrivateKey instance to decrypt with
 : param ciphertext :
 A byte string of the data to decrypt
 : param rsa _ oaep _ padding :
 If OAEP padding should be used instead of PKCS # 1 v1.5
 : 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 byte string of the plaintext"""

implementation:
if not isinstance ( private_key , PrivateKey ) :
    raise TypeError ( pretty_message ( '''
            private_key must be an instance of the PrivateKey class, not %s
            ''' , type_name ( private_key ) ) )
if not isinstance ( ciphertext , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            ciphertext must be a byte string, not %s
            ''' , type_name ( ciphertext ) ) )
if not isinstance ( rsa_oaep_padding , bool ) :
    raise TypeError ( pretty_message ( '''
            rsa_oaep_padding must be a bool, not %s
            ''' , type_name ( rsa_oaep_padding ) ) )
if _backend == 'winlegacy' :
    return _advapi32_decrypt ( private_key , ciphertext , rsa_oaep_padding )
return _bcrypt_decrypt ( private_key , ciphertext , rsa_oaep_padding )


signature:
def _advapi32_decrypt ( private_key , ciphertext , rsa_oaep_padding = False ) :
    """Encrypts a value using an RSA private key via CryptoAPI
 : param private _ key :
 A PrivateKey instance to decrypt with
 : param ciphertext :
 A byte string of the data to decrypt
 : param rsa _ oaep _ padding :
 If OAEP padding should be used instead of PKCS # 1 v1.5
 : 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 byte string of the plaintext"""

implementation:
flags = 0
if rsa_oaep_padding :
    flags = Advapi32Const . CRYPT_OAEP
ciphertext = ciphertext [ : : - 1 ]
buffer = buffer_from_bytes ( ciphertext )
out_len = new ( advapi32 , 'DWORD *' , len ( ciphertext ) )
res = advapi32 . CryptDecrypt ( private_key . ex_key_handle , null ( ) , True , flags , buffer , out_len )
handle_error ( res )
return bytes_from_buffer ( buffer , deref ( out_len ) )


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

implementation:
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


signature:
def _obtain_credentials ( self ) :
    """Obtains a credentials handle from secur32 . dll for use with SChannel"""

implementation:
protocol_values = { 'SSLv3' : Secur32Const . SP_PROT_SSL3_CLIENT , 'TLSv1' : Secur32Const . SP_PROT_TLS1_CLIENT , 'TLSv1.1' : Secur32Const . SP_PROT_TLS1_1_CLIENT , 'TLSv1.2' : Secur32Const . SP_PROT_TLS1_2_CLIENT , }
protocol_bit_mask = 0
for key , value in protocol_values . items ( ) :
    if key in self . _protocols :
        protocol_bit_mask |= value
algs = [ Secur32Const . CALG_AES_128 , Secur32Const . CALG_AES_256 , Secur32Const . CALG_3DES , Secur32Const . CALG_SHA1 , Secur32Const . CALG_ECDHE , Secur32Const . CALG_DH_EPHEM , Secur32Const . CALG_RSA_KEYX , Secur32Const . CALG_RSA_SIGN , Secur32Const . CALG_ECDSA , Secur32Const . CALG_DSS_SIGN , ]
if 'TLSv1.2' in self . _protocols :
    algs . extend ( [ Secur32Const . CALG_SHA512 , Secur32Const . CALG_SHA384 , Secur32Const . CALG_SHA256 , ] )
alg_array = new ( secur32 , 'ALG_ID[%s]' % len ( algs ) )
for index , alg in enumerate ( algs ) :
    alg_array [ index ] = alg
flags = Secur32Const . SCH_USE_STRONG_CRYPTO | Secur32Const . SCH_CRED_NO_DEFAULT_CREDS
if not self . _manual_validation and not self . _extra_trust_roots :
    flags |= Secur32Const . SCH_CRED_AUTO_CRED_VALIDATION
else :
    flags |= Secur32Const . SCH_CRED_MANUAL_CRED_VALIDATION
schannel_cred_pointer = struct ( secur32 , 'SCHANNEL_CRED' )
schannel_cred = unwrap ( schannel_cred_pointer )
schannel_cred . dwVersion = Secur32Const . SCHANNEL_CRED_VERSION
schannel_cred . cCreds = 0
schannel_cred . paCred = null ( )
schannel_cred . hRootStore = null ( )
schannel_cred . cMappers = 0
schannel_cred . aphMappers = null ( )
schannel_cred . cSupportedAlgs = len ( alg_array )
schannel_cred . palgSupportedAlgs = alg_array
schannel_cred . grbitEnabledProtocols = protocol_bit_mask
schannel_cred . dwMinimumCipherStrength = 0
schannel_cred . dwMaximumCipherStrength = 0
# Default session lifetime is 10 hours
schannel_cred . dwSessionLifespan = 0
schannel_cred . dwFlags = flags
schannel_cred . dwCredFormat = 0
cred_handle_pointer = new ( secur32 , 'CredHandle *' )
result = secur32 . AcquireCredentialsHandleW ( null ( ) , Secur32Const . UNISP_NAME , Secur32Const . SECPKG_CRED_OUTBOUND , null ( ) , schannel_cred_pointer , null ( ) , null ( ) , cred_handle_pointer , null ( ) )
handle_error ( result )
self . _credentials_handle = cred_handle_pointer


signature:
def wrap ( cls , socket , hostname , session = None ) :
    """Takes an existing socket and adds TLS
 : param socket :
 A socket . socket object to wrap with TLS
 : param hostname :
 A unicode string of the hostname or IP the socket is connected to
 : param session :
 An existing TLSSession object to allow for session reuse , specific
 protocol or manual certificate validation
 : 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"""

implementation:
if not isinstance ( socket , socket_ . socket ) :
    raise TypeError ( pretty_message ( '''
                socket must be an instance of socket.socket, not %s
                ''' , type_name ( socket ) ) )
if not isinstance ( hostname , str_cls ) :
    raise TypeError ( pretty_message ( '''
                hostname must be a unicode string, not %s
                ''' , type_name ( hostname ) ) )
if session is not None and not isinstance ( session , TLSSession ) :
    raise TypeError ( pretty_message ( '''
                session must be an instance of oscrypto.tls.TLSSession, not %s
                ''' , type_name ( session ) ) )
new_socket = cls ( None , None , session = session )
new_socket . _socket = socket
new_socket . _hostname = hostname
# Since we don ' t create the socket connection here , we can ' t try to
# reconnect with a lower version of the TLS protocol , so we just
# move the data to public exception type TLSVerificationError ( )
try :
    new_socket . _handshake ( )
except ( _TLSDowngradeError ) as e :
    new_e = TLSVerificationError ( e . message , e . certificate )
    raise new_e
except ( _TLSRetryError ) as e :
    new_e = TLSError ( e . message )
    raise new_e
return new_socket


signature:
def _create_buffers ( self , number ) :
    """Creates a SecBufferDesc struct and contained SecBuffer structs
 : param number :
 The number of contains SecBuffer objects to create
 : return :
 A tuple of ( SecBufferDesc pointer , SecBuffer array )"""

implementation:
buffers = new ( secur32 , 'SecBuffer[%d]' % number )
for index in range ( 0 , number ) :
    buffers [ index ] . cbBuffer = 0
    buffers [ index ] . BufferType = Secur32Const . SECBUFFER_EMPTY
    buffers [ index ] . pvBuffer = null ( )
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 )


signature:
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"""

implementation:
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 = x509 . Certificate . load ( last_element_cert_data )
        if last_cert . sha256 in cert_hashes :
            cert_chain_policy_para_flags |= Crypt32Const . CERT_CHAIN_POLICY_ALLOW_UNKNOWN_CA_FLAG
    ssl_extra_cert_chain_policy_para_pointer = struct ( crypt32 , 'SSL_EXTRA_CERT_CHAIN_POLICY_PARA' )
    ssl_extra_cert_chain_policy_para = unwrap ( ssl_extra_cert_chain_policy_para_pointer )
    ssl_extra_cert_chain_policy_para . cbSize = sizeof ( crypt32 , ssl_extra_cert_chain_policy_para )
    ssl_extra_cert_chain_policy_para . dwAuthType = Crypt32Const . AUTHTYPE_SERVER
    ssl_extra_cert_chain_policy_para . fdwChecks = 0
    ssl_extra_cert_chain_policy_para . pwszServerName = cast ( crypt32 , 'wchar_t *' , buffer_from_unicode ( self . _hostname ) )
    cert_chain_policy_para_pointer = struct ( crypt32 , 'CERT_CHAIN_POLICY_PARA' )
    cert_chain_policy_para = unwrap ( cert_chain_policy_para_pointer )
    cert_chain_policy_para . cbSize = sizeof ( crypt32 , cert_chain_policy_para )
    cert_chain_policy_para . dwFlags = cert_chain_policy_para_flags
    cert_chain_policy_para . pvExtraPolicyPara = cast ( crypt32 , 'void *' , ssl_extra_cert_chain_policy_para_pointer )
    cert_chain_policy_status_pointer = struct ( crypt32 , 'CERT_CHAIN_POLICY_STATUS' )
    cert_chain_policy_status = unwrap ( cert_chain_policy_status_pointer )
    cert_chain_policy_status . cbSize = sizeof ( crypt32 , cert_chain_policy_status )
    result = crypt32 . CertVerifyCertificateChainPolicy ( Crypt32Const . CERT_CHAIN_POLICY_SSL , cert_chain_context_pointer , cert_chain_policy_para_pointer , cert_chain_policy_status_pointer )
    handle_crypt32_error ( result )
    cert_context = unwrap ( cert_context_pointer )
    cert_data = bytes_from_buffer ( cert_context . pbCertEncoded , native ( int , cert_context . cbCertEncoded ) )
    cert = x509 . Certificate . load ( cert_data )
    error = cert_chain_policy_status . dwError
    if error :
        if error == Crypt32Const . CERT_E_EXPIRED :
            raise_expired_not_yet_valid ( cert )
        if error == Crypt32Const . CERT_E_UNTRUSTEDROOT :
            oscrypto_cert = load_certificate ( cert )
            if oscrypto_cert . self_signed :
                raise_self_signed ( cert )
            else :
                raise_no_issuer ( cert )
        if error == Crypt32Const . CERT_E_CN_NO_MATCH :
            raise_hostname ( cert , self . _hostname )
        if error == Crypt32Const . TRUST_E_CERT_SIGNATURE :
            raise_weak_signature ( cert )
        if error == Crypt32Const . CRYPT_E_REVOKED :
            raise_revoked ( cert )
        raise_verification ( cert )
    if cert . hash_algo in set ( [ 'md5' , 'md2' ] ) :
        raise_weak_signature ( cert )
finally :
    if store :
        crypt32 . CertCloseStore ( store , 0 )
    if cert_chain_context_pointer :
        crypt32 . CertFreeCertificateChain ( cert_chain_context_pointer )


signature:
def _handshake ( self , renegotiate = False ) :
    """Perform an initial TLS handshake , or a renegotiation
 : param renegotiate :
 If the handshake is for a renegotiation"""

implementation:
in_buffers = None
out_buffers = None
new_context_handle_pointer = None
try :
    if renegotiate :
        temp_context_handle_pointer = self . _context_handle_pointer
    else :
        new_context_handle_pointer = new ( secur32 , 'CtxtHandle *' )
        temp_context_handle_pointer = new_context_handle_pointer
    requested_flags = { Secur32Const . ISC_REQ_REPLAY_DETECT : 'replay detection' , Secur32Const . ISC_REQ_SEQUENCE_DETECT : 'sequence detection' , Secur32Const . ISC_REQ_CONFIDENTIALITY : 'confidentiality' , Secur32Const . ISC_REQ_ALLOCATE_MEMORY : 'memory allocation' , Secur32Const . ISC_REQ_INTEGRITY : 'integrity' , Secur32Const . ISC_REQ_STREAM : 'stream orientation' , Secur32Const . ISC_REQ_USE_SUPPLIED_CREDS : 'disable automatic client auth' , }
    self . _context_flags = 0
    for flag in requested_flags :
        self . _context_flags |= flag
    in_sec_buffer_desc_pointer , in_buffers = self . _create_buffers ( 2 )
    in_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_TOKEN
    out_sec_buffer_desc_pointer , out_buffers = self . _create_buffers ( 2 )
    out_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_TOKEN
    out_buffers [ 1 ] . BufferType = Secur32Const . SECBUFFER_ALERT
    output_context_flags_pointer = new ( secur32 , 'ULONG *' )
    if renegotiate :
        first_handle = temp_context_handle_pointer
        second_handle = null ( )
    else :
        first_handle = null ( )
        second_handle = temp_context_handle_pointer
    result = secur32 . InitializeSecurityContextW ( self . _session . _credentials_handle , first_handle , self . _hostname , self . _context_flags , 0 , 0 , null ( ) , 0 , second_handle , out_sec_buffer_desc_pointer , output_context_flags_pointer , null ( ) )
    if result not in set ( [ Secur32Const . SEC_E_OK , Secur32Const . SEC_I_CONTINUE_NEEDED ] ) :
        handle_error ( result , TLSError )
    if not renegotiate :
        temp_context_handle_pointer = second_handle
    else :
        temp_context_handle_pointer = first_handle
    handshake_server_bytes = b''
    handshake_client_bytes = b''
    if out_buffers [ 0 ] . cbBuffer > 0 :
        token = bytes_from_buffer ( out_buffers [ 0 ] . pvBuffer , out_buffers [ 0 ] . cbBuffer )
        handshake_client_bytes += token
        self . _socket . send ( token )
        out_buffers [ 0 ] . cbBuffer = 0
        secur32 . FreeContextBuffer ( out_buffers [ 0 ] . pvBuffer )
        out_buffers [ 0 ] . pvBuffer = null ( )
    in_data_buffer = buffer_from_bytes ( 32768 )
    in_buffers [ 0 ] . pvBuffer = cast ( secur32 , 'BYTE *' , in_data_buffer )
    bytes_read = b''
    while result != Secur32Const . SEC_E_OK :
        try :
            fail_late = False
            bytes_read = self . _socket . recv ( 8192 )
            if bytes_read == b'' :
                raise_disconnection ( )
        except ( socket_error_cls ) :
            fail_late = True
        handshake_server_bytes += bytes_read
        self . _received_bytes += bytes_read
        in_buffers [ 0 ] . cbBuffer = len ( self . _received_bytes )
        write_to_buffer ( in_data_buffer , self . _received_bytes )
        result = secur32 . InitializeSecurityContextW ( self . _session . _credentials_handle , temp_context_handle_pointer , self . _hostname , self . _context_flags , 0 , 0 , in_sec_buffer_desc_pointer , 0 , null ( ) , out_sec_buffer_desc_pointer , output_context_flags_pointer , null ( ) )
        if result == Secur32Const . SEC_E_INCOMPLETE_MESSAGE :
            in_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_TOKEN
            # Windows 10 seems to fill the second input buffer with
            # a BufferType of SECBUFFER _ MISSING ( 4 ) , which if not
            # cleared causes the handshake to fail .
            if in_buffers [ 1 ] . BufferType != Secur32Const . SECBUFFER_EMPTY :
                in_buffers [ 1 ] . BufferType = Secur32Const . SECBUFFER_EMPTY
                in_buffers [ 1 ] . cbBuffer = 0
                if not is_null ( in_buffers [ 1 ] . pvBuffer ) :
                    secur32 . FreeContextBuffer ( in_buffers [ 1 ] . pvBuffer )
                    in_buffers [ 1 ] . pvBuffer = null ( )
            if fail_late :
                raise_disconnection ( )
            continue
        if result == Secur32Const . SEC_E_ILLEGAL_MESSAGE :
            if detect_client_auth_request ( handshake_server_bytes ) :
                raise_client_auth ( )
            alert_info = parse_alert ( handshake_server_bytes )
            if alert_info and alert_info == ( 2 , 70 ) :
                raise_protocol_version ( )
            raise_handshake ( )
        if result == Secur32Const . SEC_E_WRONG_PRINCIPAL :
            chain = extract_chain ( handshake_server_bytes )
            raise_hostname ( chain [ 0 ] , self . _hostname )
        if result == Secur32Const . SEC_E_CERT_EXPIRED :
            chain = extract_chain ( handshake_server_bytes )
            raise_expired_not_yet_valid ( chain [ 0 ] )
        if result == Secur32Const . SEC_E_UNTRUSTED_ROOT :
            chain = extract_chain ( handshake_server_bytes )
            cert = chain [ 0 ]
            oscrypto_cert = load_certificate ( cert )
            if not oscrypto_cert . self_signed :
                raise_no_issuer ( cert )
            raise_self_signed ( cert )
        if result == Secur32Const . SEC_E_INTERNAL_ERROR :
            if get_dh_params_length ( handshake_server_bytes ) < 1024 :
                raise_dh_params ( )
        if result == Secur32Const . SEC_I_INCOMPLETE_CREDENTIALS :
            raise_client_auth ( )
        if result == Crypt32Const . TRUST_E_CERT_SIGNATURE :
            raise_weak_signature ( cert )
        if result == Secur32Const . SEC_E_INVALID_TOKEN : # If an alert it present , there may have been a handshake
        # error due to the server using a certificate path with a
        # trust root using MD2 or MD5 combined with TLS 1.2 . To
        # work around this , if the user allows anything other than
        # TLS 1.2 , we just remove it from the acceptable protocols
        # and try again .
            if out_buffers [ 1 ] . cbBuffer > 0 :
                alert_bytes = bytes_from_buffer ( out_buffers [ 1 ] . pvBuffer , out_buffers [ 1 ] . cbBuffer )
                handshake_client_bytes += alert_bytes
                alert_number = alert_bytes [ 6 : 7 ]
                if alert_number == b'\x28' or alert_number == b'\x2b' :
                    if 'TLSv1.2' in self . _session . _protocols and len ( self . _session . _protocols ) > 1 :
                        chain = extract_chain ( handshake_server_bytes )
                        raise _TLSDowngradeError ( 'Server certificate verification failed - weak certificate signature algorithm' , chain [ 0 ] )
            if detect_client_auth_request ( handshake_server_bytes ) :
                raise_client_auth ( )
            if detect_other_protocol ( handshake_server_bytes ) :
                raise_protocol_error ( handshake_server_bytes )
            raise_handshake ( )
        # These are semi - common errors with TLSv1.2 on Windows 7 an 8
        # that appears to be due to poor handling of the
        # ServerKeyExchange for DHE _ RSA cipher suites . The solution
        # is to retry the handshake .
        if result == Secur32Const . SEC_E_BUFFER_TOO_SMALL or result == Secur32Const . SEC_E_MESSAGE_ALTERED :
            if 'TLSv1.2' in self . _session . _protocols :
                raise _TLSRetryError ( 'TLS handshake failed' )
        if fail_late :
            raise_disconnection ( )
        if result == Secur32Const . SEC_E_INVALID_PARAMETER :
            if get_dh_params_length ( handshake_server_bytes ) < 1024 :
                raise_dh_params ( )
        if result not in set ( [ Secur32Const . SEC_E_OK , Secur32Const . SEC_I_CONTINUE_NEEDED ] ) :
            handle_error ( result , TLSError )
        if out_buffers [ 0 ] . cbBuffer > 0 :
            token = bytes_from_buffer ( out_buffers [ 0 ] . pvBuffer , out_buffers [ 0 ] . cbBuffer )
            handshake_client_bytes += token
            self . _socket . send ( token )
            out_buffers [ 0 ] . cbBuffer = 0
            secur32 . FreeContextBuffer ( out_buffers [ 0 ] . pvBuffer )
            out_buffers [ 0 ] . pvBuffer = null ( )
        if in_buffers [ 1 ] . BufferType == Secur32Const . SECBUFFER_EXTRA :
            extra_amount = in_buffers [ 1 ] . cbBuffer
            self . _received_bytes = self . _received_bytes [ - extra_amount : ]
            in_buffers [ 1 ] . BufferType = Secur32Const . SECBUFFER_EMPTY
            in_buffers [ 1 ] . cbBuffer = 0
            secur32 . FreeContextBuffer ( in_buffers [ 1 ] . pvBuffer )
            in_buffers [ 1 ] . pvBuffer = null ( )
            # The handshake is complete , so discard any extra bytes
            if result == Secur32Const . SEC_E_OK :
                handshake_server_bytes = handshake_server_bytes [ - extra_amount : ]
        else :
            self . _received_bytes = b''
    connection_info_pointer = struct ( secur32 , 'SecPkgContext_ConnectionInfo' )
    result = secur32 . QueryContextAttributesW ( temp_context_handle_pointer , Secur32Const . SECPKG_ATTR_CONNECTION_INFO , connection_info_pointer )
    handle_error ( result , TLSError )
    connection_info = unwrap ( connection_info_pointer )
    self . _protocol = { Secur32Const . SP_PROT_SSL2_CLIENT : 'SSLv2' , Secur32Const . SP_PROT_SSL3_CLIENT : 'SSLv3' , Secur32Const . SP_PROT_TLS1_CLIENT : 'TLSv1' , Secur32Const . SP_PROT_TLS1_1_CLIENT : 'TLSv1.1' , Secur32Const . SP_PROT_TLS1_2_CLIENT : 'TLSv1.2' , } . get ( native ( int , connection_info . dwProtocol ) , str_cls ( connection_info . dwProtocol ) )
    if self . _protocol in set ( [ 'SSLv3' , 'TLSv1' , 'TLSv1.1' , 'TLSv1.2' ] ) :
        session_info = parse_session_info ( handshake_server_bytes , handshake_client_bytes )
        self . _cipher_suite = session_info [ 'cipher_suite' ]
        self . _compression = session_info [ 'compression' ]
        self . _session_id = session_info [ 'session_id' ]
        self . _session_ticket = session_info [ 'session_ticket' ]
    output_context_flags = deref ( output_context_flags_pointer )
    for flag in requested_flags :
        if ( flag | output_context_flags ) == 0 :
            raise OSError ( pretty_message ( '''
                        Unable to obtain a credential context with the property %s
                        ''' , requested_flags [ flag ] ) )
    if not renegotiate :
        self . _context_handle_pointer = temp_context_handle_pointer
        new_context_handle_pointer = None
        stream_sizes_pointer = struct ( secur32 , 'SecPkgContext_StreamSizes' )
        result = secur32 . QueryContextAttributesW ( self . _context_handle_pointer , Secur32Const . SECPKG_ATTR_STREAM_SIZES , stream_sizes_pointer )
        handle_error ( result )
        stream_sizes = unwrap ( stream_sizes_pointer )
        self . _header_size = native ( int , stream_sizes . cbHeader )
        self . _message_size = native ( int , stream_sizes . cbMaximumMessage )
        self . _trailer_size = native ( int , stream_sizes . cbTrailer )
        self . _buffer_size = self . _header_size + self . _message_size + self . _trailer_size
    if self . _session . _extra_trust_roots :
        self . _extra_trust_root_validation ( )
except ( OSError , socket_ . error ) :
    self . close ( )
    raise
finally :
    if out_buffers :
        if not is_null ( out_buffers [ 0 ] . pvBuffer ) :
            secur32 . FreeContextBuffer ( out_buffers [ 0 ] . pvBuffer )
        if not is_null ( out_buffers [ 1 ] . pvBuffer ) :
            secur32 . FreeContextBuffer ( out_buffers [ 1 ] . pvBuffer )
    if new_context_handle_pointer :
        secur32 . DeleteSecurityContext ( new_context_handle_pointer )


signature:
def read ( self , max_length ) :
    """Reads data from the TLS - wrapped socket
 : param max _ length :
 The number of bytes to read
 : raises :
 socket . socket - when a non - TLS socket error occurs
 oscrypto . errors . TLSError - when a TLS - related error occurs
 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 byte string of the data read"""

implementation:
if not isinstance ( max_length , int_types ) :
    raise TypeError ( pretty_message ( '''
                max_length must be an integer, not %s
                ''' , type_name ( max_length ) ) )
if self . _context_handle_pointer is None : # Allow the user to read any remaining decrypted data
    if self . _decrypted_bytes != b'' :
        output = self . _decrypted_bytes [ 0 : max_length ]
        self . _decrypted_bytes = self . _decrypted_bytes [ max_length : ]
        return output
    self . _raise_closed ( )
# The first time read is called , set up a single contiguous buffer that
# it used by DecryptMessage ( ) to populate the three output buffers .
# Since we are creating the buffer , we do not need to free it other
# than allowing Python to GC it once this object is GCed .
if not self . _decrypt_data_buffer :
    self . _decrypt_data_buffer = buffer_from_bytes ( self . _buffer_size )
    self . _decrypt_desc , self . _decrypt_buffers = self . _create_buffers ( 4 )
    self . _decrypt_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_DATA
    self . _decrypt_buffers [ 0 ] . pvBuffer = cast ( secur32 , 'BYTE *' , self . _decrypt_data_buffer )
to_recv = max ( max_length , self . _buffer_size )
# These variables are set to reduce dict access and function calls
# in the read loop . Also makes the code easier to read .
null_value = null ( )
buf0 = self . _decrypt_buffers [ 0 ]
buf1 = self . _decrypt_buffers [ 1 ]
buf2 = self . _decrypt_buffers [ 2 ]
buf3 = self . _decrypt_buffers [ 3 ]
def _reset_buffers ( ) :
    buf0 . BufferType = Secur32Const . SECBUFFER_DATA
    buf0 . pvBuffer = cast ( secur32 , 'BYTE *' , self . _decrypt_data_buffer )
    buf0 . cbBuffer = 0
    buf1 . BufferType = Secur32Const . SECBUFFER_EMPTY
    buf1 . pvBuffer = null_value
    buf1 . cbBuffer = 0
    buf2 . BufferType = Secur32Const . SECBUFFER_EMPTY
    buf2 . pvBuffer = null_value
    buf2 . cbBuffer = 0
    buf3 . BufferType = Secur32Const . SECBUFFER_EMPTY
    buf3 . pvBuffer = null_value
    buf3 . cbBuffer = 0
output = self . _decrypted_bytes
output_len = len ( output )
self . _decrypted_bytes = b''
# Don ' t block if we have buffered data available
if output_len > 0 and not self . select_read ( 0 ) :
    self . _decrypted_bytes = b''
    return output
# This read loop will only be run if there wasn ' t enough
# buffered data to fulfill the requested max _ length
do_read = len ( self . _received_bytes ) == 0
while output_len < max_length :
    if do_read :
        self . _received_bytes += self . _socket . recv ( to_recv )
        if len ( self . _received_bytes ) == 0 :
            raise_disconnection ( )
    data_len = min ( len ( self . _received_bytes ) , self . _buffer_size )
    if data_len == 0 :
        break
    self . _decrypt_buffers [ 0 ] . cbBuffer = data_len
    write_to_buffer ( self . _decrypt_data_buffer , self . _received_bytes [ 0 : data_len ] )
    result = secur32 . DecryptMessage ( self . _context_handle_pointer , self . _decrypt_desc , 0 , null ( ) )
    do_read = False
    if result == Secur32Const . SEC_E_INCOMPLETE_MESSAGE :
        _reset_buffers ( )
        do_read = True
        continue
    elif result == Secur32Const . SEC_I_CONTEXT_EXPIRED :
        self . _remote_closed = True
        self . shutdown ( )
        break
    elif result == Secur32Const . SEC_I_RENEGOTIATE :
        self . _handshake ( renegotiate = True )
        return self . read ( max_length )
    elif result != Secur32Const . SEC_E_OK :
        handle_error ( result , TLSError )
    valid_buffer_types = set ( [ Secur32Const . SECBUFFER_EMPTY , Secur32Const . SECBUFFER_STREAM_HEADER , Secur32Const . SECBUFFER_STREAM_TRAILER ] )
    extra_amount = None
    for buf in ( buf0 , buf1 , buf2 , buf3 ) :
        buffer_type = buf . BufferType
        if buffer_type == Secur32Const . SECBUFFER_DATA :
            output += bytes_from_buffer ( buf . pvBuffer , buf . cbBuffer )
            output_len = len ( output )
        elif buffer_type == Secur32Const . SECBUFFER_EXTRA :
            extra_amount = native ( int , buf . cbBuffer )
        elif buffer_type not in valid_buffer_types :
            raise OSError ( pretty_message ( '''
                        Unexpected decrypt output buffer of type %s
                        ''' , buffer_type ) )
    if extra_amount :
        self . _received_bytes = self . _received_bytes [ data_len - extra_amount : ]
    else :
        self . _received_bytes = self . _received_bytes [ data_len : ]
    # Here we reset the structs for the next call to DecryptMessage ( )
    _reset_buffers ( )
    # If we have read something , but there is nothing left to read , we
    # break so that we don ' t block for longer than necessary
    if self . select_read ( 0 ) :
        do_read = True
    if not do_read and len ( self . _received_bytes ) == 0 :
        break
# If the output is more than we requested ( because data is decrypted in
# blocks ) , we save the extra in a buffer
if len ( output ) > max_length :
    self . _decrypted_bytes = output [ max_length : ]
    output = output [ 0 : max_length ]
return output


signature:
def select_read ( self , timeout = None ) :
    """Blocks until the socket is ready to be read from , or the timeout is hit
 : param timeout :
 A float - the period of time to wait for data to be read . None for
 no time limit .
 : return :
 A boolean - if data is ready to be read . Will only be False if
 timeout is not None ."""

implementation:
# If we have buffered data , we consider a read possible
if len ( self . _decrypted_bytes ) > 0 :
    return True
read_ready , _ , _ = select . select ( [ self . _socket ] , [ ] , [ ] , timeout )
return len ( read_ready ) > 0


signature:
def read_exactly ( self , num_bytes ) :
    """Reads exactly the specified number of bytes from the socket
 : param num _ bytes :
 An integer - the exact number of bytes to read
 : return :
 A byte string of the data that was read"""

implementation:
output = b''
remaining = num_bytes
while remaining > 0 :
    output += self . read ( remaining )
    remaining = num_bytes - len ( output )
return output


signature:
def write ( self , data ) :
    """Writes data to the TLS - wrapped socket
 : param data :
 A byte string to write to the socket
 : raises :
 socket . socket - when a non - TLS socket error occurs
 oscrypto . errors . TLSError - when a TLS - related error occurs
 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"""

implementation:
if self . _context_handle_pointer is None :
    self . _raise_closed ( )
if not self . _encrypt_data_buffer :
    self . _encrypt_data_buffer = buffer_from_bytes ( self . _header_size + self . _message_size + self . _trailer_size )
    self . _encrypt_desc , self . _encrypt_buffers = self . _create_buffers ( 4 )
    self . _encrypt_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_STREAM_HEADER
    self . _encrypt_buffers [ 0 ] . cbBuffer = self . _header_size
    self . _encrypt_buffers [ 0 ] . pvBuffer = cast ( secur32 , 'BYTE *' , self . _encrypt_data_buffer )
    self . _encrypt_buffers [ 1 ] . BufferType = Secur32Const . SECBUFFER_DATA
    self . _encrypt_buffers [ 1 ] . pvBuffer = ref ( self . _encrypt_data_buffer , self . _header_size )
    self . _encrypt_buffers [ 2 ] . BufferType = Secur32Const . SECBUFFER_STREAM_TRAILER
    self . _encrypt_buffers [ 2 ] . cbBuffer = self . _trailer_size
    self . _encrypt_buffers [ 2 ] . pvBuffer = ref ( self . _encrypt_data_buffer , self . _header_size + self . _message_size )
while len ( data ) > 0 :
    to_write = min ( len ( data ) , self . _message_size )
    write_to_buffer ( self . _encrypt_data_buffer , data [ 0 : to_write ] , self . _header_size )
    self . _encrypt_buffers [ 1 ] . cbBuffer = to_write
    self . _encrypt_buffers [ 2 ] . pvBuffer = ref ( self . _encrypt_data_buffer , self . _header_size + to_write )
    result = secur32 . EncryptMessage ( self . _context_handle_pointer , 0 , self . _encrypt_desc , 0 )
    if result != Secur32Const . SEC_E_OK :
        handle_error ( result , TLSError )
    to_send = native ( int , self . _encrypt_buffers [ 0 ] . cbBuffer )
    to_send += native ( int , self . _encrypt_buffers [ 1 ] . cbBuffer )
    to_send += native ( int , self . _encrypt_buffers [ 2 ] . cbBuffer )
    try :
        self . _socket . send ( bytes_from_buffer ( self . _encrypt_data_buffer , to_send ) )
    except ( socket_ . error ) as e :
        if e . errno == 10053 :
            raise_disconnection ( )
        raise
    data = data [ to_send : ]


signature:
def select_write ( self , timeout = None ) :
    """Blocks until the socket is ready to be written to , or the timeout is hit
 : param timeout :
 A float - the period of time to wait for the socket to be ready to
 written to . None for no time limit .
 : return :
 A boolean - if the socket is ready for writing . Will only be False
 if timeout is not None ."""

implementation:
_ , write_ready , _ = select . select ( [ ] , [ self . _socket ] , [ ] , timeout )
return len ( write_ready ) > 0


signature:
def shutdown ( self ) :
    """Shuts down the TLS session and then shuts down the underlying socket
 : raises :
 OSError - when an error is returned by the OS crypto library"""

implementation:
if self . _context_handle_pointer is None :
    return
out_buffers = None
try : # ApplyControlToken fails with SEC _ E _ UNSUPPORTED _ FUNCTION
# when called on Windows 7
    if _win_version_info >= ( 6 , 2 ) :
        buffers = new ( secur32 , 'SecBuffer[1]' )
        # This is a SCHANNEL _ SHUTDOWN token ( DWORD of 1)
        buffers [ 0 ] . cbBuffer = 4
        buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_TOKEN
        buffers [ 0 ] . pvBuffer = cast ( secur32 , 'BYTE *' , buffer_from_bytes ( b'\x01\x00\x00\x00' ) )
        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 = 1
        sec_buffer_desc . pBuffers = buffers
        result = secur32 . ApplyControlToken ( self . _context_handle_pointer , sec_buffer_desc_pointer )
        handle_error ( result , TLSError )
    out_sec_buffer_desc_pointer , out_buffers = self . _create_buffers ( 2 )
    out_buffers [ 0 ] . BufferType = Secur32Const . SECBUFFER_TOKEN
    out_buffers [ 1 ] . BufferType = Secur32Const . SECBUFFER_ALERT
    output_context_flags_pointer = new ( secur32 , 'ULONG *' )
    result = secur32 . InitializeSecurityContextW ( self . _session . _credentials_handle , self . _context_handle_pointer , self . _hostname , self . _context_flags , 0 , 0 , null ( ) , 0 , null ( ) , out_sec_buffer_desc_pointer , output_context_flags_pointer , null ( ) )
    acceptable_results = set ( [ Secur32Const . SEC_E_OK , Secur32Const . SEC_E_CONTEXT_EXPIRED , Secur32Const . SEC_I_CONTINUE_NEEDED ] )
    if result not in acceptable_results :
        handle_error ( result , TLSError )
    token = bytes_from_buffer ( out_buffers [ 0 ] . pvBuffer , out_buffers [ 0 ] . cbBuffer )
    try : # If there is an error sending the shutdown , ignore it since the
    # connection is likely gone at this point
        self . _socket . send ( token )
    except ( socket_ . error ) :
        pass
finally :
    if out_buffers :
        if not is_null ( out_buffers [ 0 ] . pvBuffer ) :
            secur32 . FreeContextBuffer ( out_buffers [ 0 ] . pvBuffer )
        if not is_null ( out_buffers [ 1 ] . pvBuffer ) :
            secur32 . FreeContextBuffer ( out_buffers [ 1 ] . pvBuffer )
    secur32 . DeleteSecurityContext ( self . _context_handle_pointer )
    self . _context_handle_pointer = None
    try :
        self . _socket . shutdown ( socket_ . SHUT_RDWR )
    except ( socket_ . error ) :
        pass


signature:
def close ( self ) :
    """Shuts down the TLS session and socket and forcibly closes it"""

implementation:
try :
    self . shutdown ( )
finally :
    if self . _socket :
        try :
            self . _socket . close ( )
        except ( socket_ . error ) :
            pass
        self . _socket = None


signature:
def _read_certificates ( self ) :
    """Reads end - entity and intermediate certificate information from the
 TLS session"""

implementation:
cert_context_pointer_pointer = new ( crypt32 , 'CERT_CONTEXT **' )
result = secur32 . QueryContextAttributesW ( self . _context_handle_pointer , Secur32Const . SECPKG_ATTR_REMOTE_CERT_CONTEXT , cert_context_pointer_pointer )
handle_error ( result , TLSError )
cert_context_pointer = unwrap ( cert_context_pointer_pointer )
cert_context_pointer = cast ( crypt32 , 'CERT_CONTEXT *' , cert_context_pointer )
cert_context = unwrap ( cert_context_pointer )
cert_data = bytes_from_buffer ( cert_context . pbCertEncoded , native ( int , cert_context . cbCertEncoded ) )
self . _certificate = x509 . Certificate . load ( cert_data )
self . _intermediates = [ ]
store_handle = None
try :
    store_handle = cert_context . hCertStore
    context_pointer = crypt32 . CertEnumCertificatesInStore ( store_handle , null ( ) )
    while not is_null ( context_pointer ) :
        context = unwrap ( context_pointer )
        data = bytes_from_buffer ( context . pbCertEncoded , native ( int , context . cbCertEncoded ) )
        # The cert store seems to include the end - entity certificate as
        # the last entry , but we already have that from the struct .
        if data != cert_data :
            self . _intermediates . append ( x509 . Certificate . load ( data ) )
        context_pointer = crypt32 . CertEnumCertificatesInStore ( store_handle , context_pointer )
finally :
    if store_handle :
        crypt32 . CertCloseStore ( store_handle , 0 )


signature:
def certificate ( self ) :
    """An asn1crypto . x509 . Certificate object of the end - entity certificate
 presented by the server"""

implementation:
if self . _context_handle_pointer is None :
    self . _raise_closed ( )
if self . _certificate is None :
    self . _read_certificates ( )
return self . _certificate


signature:
def intermediates ( self ) :
    """A list of asn1crypto . x509 . Certificate objects that were presented as
 intermediates by the server"""

implementation:
if self . _context_handle_pointer is None :
    self . _raise_closed ( )
if self . _certificate is None :
    self . _read_certificates ( )
return self . _intermediates


signature:
def handle_cf_error ( error_pointer ) :
    """Checks a CFErrorRef and throws an exception if there is an error to report
 : param error _ pointer :
 A CFErrorRef
 : raises :
 OSError - when the CFErrorRef contains an error"""

implementation:
if is_null ( error_pointer ) :
    return
error = unwrap ( error_pointer )
if is_null ( error ) :
    return
cf_string_domain = CoreFoundation . CFErrorGetDomain ( error )
domain = CFHelpers . cf_string_to_unicode ( cf_string_domain )
CoreFoundation . CFRelease ( cf_string_domain )
num = CoreFoundation . CFErrorGetCode ( error )
cf_string_ref = CoreFoundation . CFErrorCopyDescription ( error )
output = CFHelpers . cf_string_to_unicode ( cf_string_ref )
CoreFoundation . CFRelease ( cf_string_ref )
if output is None :
    if domain == 'NSOSStatusErrorDomain' :
        code_map = { - 2147416010 : 'ACL add failed' , - 2147416025 : 'ACL base certs not supported' , - 2147416019 : 'ACL challenge callback failed' , - 2147416015 : 'ACL change failed' , - 2147416012 : 'ACL delete failed' , - 2147416017 : 'ACL entry tag not found' , - 2147416011 : 'ACL replace failed' , - 2147416021 : 'ACL subject type not supported' , - 2147415789 : 'Algid mismatch' , - 2147415726 : 'Already logged in' , - 2147415040 : 'Apple add application ACL subject' , - 2147415036 : 'Apple invalid key end date' , - 2147415037 : 'Apple invalid key start date' , - 2147415039 : 'Apple public key incomplete' , - 2147415038 : 'Apple signature mismatch' , - 2147415034 : 'Apple SSLv2 rollback' , - 2147415802 : 'Attach handle busy' , - 2147415731 : 'Block size mismatch' , - 2147415722 : 'Crypto data callback failed' , - 2147415804 : 'Device error' , - 2147415835 : 'Device failed' , - 2147415803 : 'Device memory error' , - 2147415836 : 'Device reset' , - 2147415728 : 'Device verify failed' , - 2147416054 : 'Function failed' , - 2147416057 : 'Function not implemented' , - 2147415807 : 'Input length error' , - 2147415837 : 'Insufficient client identification' , - 2147416063 : 'Internal error' , - 2147416027 : 'Invalid access credentials' , - 2147416026 : 'Invalid ACL base certs' , - 2147416020 : 'Invalid ACL challenge callback' , - 2147416016 : 'Invalid ACL edit mode' , - 2147416018 : 'Invalid ACL entry tag' , - 2147416022 : 'Invalid ACL subject value' , - 2147415759 : 'Invalid algorithm' , - 2147415678 : 'Invalid attr access credentials' , - 2147415704 : 'Invalid attr alg params' , - 2147415686 : 'Invalid attr base' , - 2147415738 : 'Invalid attr block size' , - 2147415680 : 'Invalid attr dl db handle' , - 2147415696 : 'Invalid attr effective bits' , - 2147415692 : 'Invalid attr end date' , - 2147415752 : 'Invalid attr init vector' , - 2147415682 : 'Invalid attr iteration count' , - 2147415754 : 'Invalid attr key' , - 2147415740 : 'Invalid attr key length' , - 2147415700 : 'Invalid attr key type' , - 2147415702 : 'Invalid attr label' , - 2147415698 : 'Invalid attr mode' , - 2147415708 : 'Invalid attr output size' , - 2147415748 : 'Invalid attr padding' , - 2147415742 : 'Invalid attr passphrase' , - 2147415688 : 'Invalid attr prime' , - 2147415674 : 'Invalid attr private key format' , - 2147415676 : 'Invalid attr public key format' , - 2147415746 : 'Invalid attr random' , - 2147415706 : 'Invalid attr rounds' , - 2147415750 : 'Invalid attr salt' , - 2147415744 : 'Invalid attr seed' , - 2147415694 : 'Invalid attr start date' , - 2147415684 : 'Invalid attr subprime' , - 2147415672 : 'Invalid attr symmetric key format' , - 2147415690 : 'Invalid attr version' , - 2147415670 : 'Invalid attr wrapped key format' , - 2147415760 : 'Invalid context' , - 2147416000 : 'Invalid context handle' , - 2147415976 : 'Invalid crypto data' , - 2147415994 : 'Invalid data' , - 2147415768 : 'Invalid data count' , - 2147415723 : 'Invalid digest algorithm' , - 2147416059 : 'Invalid input pointer' , - 2147415766 : 'Invalid input vector' , - 2147415792 : 'Invalid key' , - 2147415780 : 'Invalid keyattr mask' , - 2147415782 : 'Invalid keyusage mask' , - 2147415790 : 'Invalid key class' , - 2147415776 : 'Invalid key format' , - 2147415778 : 'Invalid key label' , - 2147415783 : 'Invalid key pointer' , - 2147415791 : 'Invalid key reference' , - 2147415727 : 'Invalid login name' , - 2147416014 : 'Invalid new ACL entry' , - 2147416013 : 'Invalid new ACL owner' , - 2147416058 : 'Invalid output pointer' , - 2147415765 : 'Invalid output vector' , - 2147415978 : 'Invalid passthrough id' , - 2147416060 : 'Invalid pointer' , - 2147416024 : 'Invalid sample value' , - 2147415733 : 'Invalid signature' , - 2147415787 : 'Key blob type incorrect' , - 2147415786 : 'Key header inconsistent' , - 2147415724 : 'Key label already exists' , - 2147415788 : 'Key usage incorrect' , - 2147416061 : 'Mds error' , - 2147416062 : 'Memory error' , - 2147415677 : 'Missing attr access credentials' , - 2147415703 : 'Missing attr alg params' , - 2147415685 : 'Missing attr base' , - 2147415737 : 'Missing attr block size' , - 2147415679 : 'Missing attr dl db handle' , - 2147415695 : 'Missing attr effective bits' , - 2147415691 : 'Missing attr end date' , - 2147415751 : 'Missing attr init vector' , - 2147415681 : 'Missing attr iteration count' , - 2147415753 : 'Missing attr key' , - 2147415739 : 'Missing attr key length' , - 2147415699 : 'Missing attr key type' , - 2147415701 : 'Missing attr label' , - 2147415697 : 'Missing attr mode' , - 2147415707 : 'Missing attr output size' , - 2147415747 : 'Missing attr padding' , - 2147415741 : 'Missing attr passphrase' , - 2147415687 : 'Missing attr prime' , - 2147415673 : 'Missing attr private key format' , - 2147415675 : 'Missing attr public key format' , - 2147415745 : 'Missing attr random' , - 2147415705 : 'Missing attr rounds' , - 2147415749 : 'Missing attr salt' , - 2147415743 : 'Missing attr seed' , - 2147415693 : 'Missing attr start date' , - 2147415683 : 'Missing attr subprime' , - 2147415671 : 'Missing attr symmetric key format' , - 2147415689 : 'Missing attr version' , - 2147415669 : 'Missing attr wrapped key format' , - 2147415801 : 'Not logged in' , - 2147415840 : 'No user interaction' , - 2147416029 : 'Object ACL not supported' , - 2147416028 : 'Object ACL required' , - 2147416030 : 'Object manip auth denied' , - 2147416031 : 'Object use auth denied' , - 2147416032 : 'Operation auth denied' , - 2147416055 : 'OS access denied' , - 2147415806 : 'Output length error' , - 2147415725 : 'Private key already exists' , - 2147415730 : 'Private key not found' , - 2147415989 : 'Privilege not granted' , - 2147415805 : 'Privilege not supported' , - 2147415729 : 'Public key inconsistent' , - 2147415732 : 'Query size unknown' , - 2147416023 : 'Sample value not supported' , - 2147416056 : 'Self check failed' , - 2147415838 : 'Service not available' , - 2147415736 : 'Staged operation in progress' , - 2147415735 : 'Staged operation not started' , - 2147415779 : 'Unsupported keyattr mask' , - 2147415781 : 'Unsupported keyusage mask' , - 2147415785 : 'Unsupported key format' , - 2147415777 : 'Unsupported key label' , - 2147415784 : 'Unsupported key size' , - 2147415839 : 'User canceled' , - 2147415767 : 'Vector of bufs unsupported' , - 2147415734 : 'Verify failed' , }
        if num in code_map :
            output = code_map [ num ]
    if not output :
        output = '%s %s' % ( domain , num )
raise OSError ( output )


implementation:
if not isinstance ( socket , socket_ . socket ) :
    raise TypeError ( pretty_message ( '''
                socket must be an instance of socket.socket, not %s
                ''' , type_name ( socket ) ) )
if not isinstance ( hostname , str_cls ) :
    raise TypeError ( pretty_message ( '''
                hostname must be a unicode string, not %s
                ''' , type_name ( hostname ) ) )
if session is not None and not isinstance ( session , TLSSession ) :
    raise TypeError ( pretty_message ( '''
                session must be an instance of oscrypto.tls.TLSSession, not %s
                ''' , type_name ( session ) ) )
new_socket = cls ( None , None , session = session )
new_socket . _socket = socket
new_socket . _hostname = hostname
new_socket . _handshake ( )
return new_socket