signature:
def ecdsa_sign ( private_key , data , hash_algorithm ) :
    """Generates an ECDSA signature in pure Python ( thus slow )
 : 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 " 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 not hasattr ( private_key , 'asn1' ) or not isinstance ( private_key . asn1 , keys . PrivateKeyInfo ) :
    raise TypeError ( pretty_message ( '''
            private_key must be an instance of the
            oscrypto.asymmetric.PrivateKey class, not %s
            ''' , type_name ( private_key ) ) )
curve_name = private_key . curve
if curve_name not in set ( [ 'secp256r1' , 'secp384r1' , 'secp521r1' ] ) :
    raise ValueError ( pretty_message ( '''
            private_key does not use one of the named curves secp256r1,
            secp384r1 or secp521r1
            ''' ) )
if not isinstance ( data , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            data must be a byte string, not %s
            ''' , type_name ( data ) ) )
if hash_algorithm not in set ( [ 'sha1' , 'sha224' , 'sha256' , 'sha384' , 'sha512' ] ) :
    raise ValueError ( pretty_message ( '''
            hash_algorithm must be one of "sha1", "sha224", "sha256", "sha384",
            "sha512", not %s
            ''' , repr ( hash_algorithm ) ) )
hash_func = getattr ( hashlib , hash_algorithm )
ec_private_key = private_key . asn1 [ 'private_key' ] . parsed
private_key_bytes = ec_private_key [ 'private_key' ] . contents
private_key_int = ec_private_key [ 'private_key' ] . native
curve_num_bytes = CURVE_BYTES [ curve_name ]
curve_base_point = { 'secp256r1' : SECP256R1_BASE_POINT , 'secp384r1' : SECP384R1_BASE_POINT , 'secp521r1' : SECP521R1_BASE_POINT , } [ curve_name ]
n = curve_base_point . order
# RFC 6979 section 3.2
digest = hash_func ( data ) . digest ( )
hash_length = len ( digest )
h = int_from_bytes ( digest , signed = False ) % n
V = b'\x01' * hash_length
K = b'\x00' * hash_length
K = hmac . new ( K , V + b'\x00' + private_key_bytes + digest , hash_func ) . digest ( )
V = hmac . new ( K , V , hash_func ) . digest ( )
K = hmac . new ( K , V + b'\x01' + private_key_bytes + digest , hash_func ) . digest ( )
V = hmac . new ( K , V , hash_func ) . digest ( )
r = 0
s = 0
while True : # h . 1
    T = b''
    # h . 2
    while len ( T ) < curve_num_bytes :
        V = hmac . new ( K , V , hash_func ) . digest ( )
        T += V
    # h . 3
    k = int_from_bytes ( T [ 0 : curve_num_bytes ] , signed = False )
    if k == 0 or k >= n :
        continue
    # Calculate the signature in the loop in case we need a new k
    r = ( curve_base_point * k ) . x % n
    if r == 0 :
        continue
    s = ( inverse_mod ( k , n ) * ( h + ( private_key_int * r ) % n ) ) % n
    if s == 0 :
        continue
    break
return DSASignature ( { 'r' : r , 's' : s } ) . dump ( )


signature:
def ecdsa_verify ( certificate_or_public_key , signature , data , hash_algorithm ) :
    """Verifies an ECDSA signature in pure Python ( thus slow )
 : 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:
has_asn1 = hasattr ( certificate_or_public_key , 'asn1' )
if not has_asn1 or not isinstance ( certificate_or_public_key . asn1 , ( keys . PublicKeyInfo , Certificate ) ) :
    raise TypeError ( pretty_message ( '''
            certificate_or_public_key must be an instance of the
            oscrypto.asymmetric.PublicKey or oscrypto.asymmetric.Certificate
            classes, not %s
            ''' , type_name ( certificate_or_public_key ) ) )
curve_name = certificate_or_public_key . curve
if curve_name not in set ( [ 'secp256r1' , 'secp384r1' , 'secp521r1' ] ) :
    raise ValueError ( pretty_message ( '''
            certificate_or_public_key does not use one of the named curves
            secp256r1, secp384r1 or secp521r1
            ''' ) )
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 ) ) )
if hash_algorithm not in set ( [ 'sha1' , 'sha224' , 'sha256' , 'sha384' , 'sha512' ] ) :
    raise ValueError ( pretty_message ( '''
            hash_algorithm must be one of "sha1", "sha224", "sha256", "sha384",
            "sha512", not %s
            ''' , repr ( hash_algorithm ) ) )
asn1 = certificate_or_public_key . asn1
if isinstance ( asn1 , Certificate ) :
    asn1 = asn1 . public_key
curve_base_point = { 'secp256r1' : SECP256R1_BASE_POINT , 'secp384r1' : SECP384R1_BASE_POINT , 'secp521r1' : SECP521R1_BASE_POINT , } [ curve_name ]
x , y = asn1 [ 'public_key' ] . to_coords ( )
n = curve_base_point . order
# Validates that the point is valid
public_key_point = PrimePoint ( curve_base_point . curve , x , y , n )
try :
    signature = DSASignature . load ( signature )
    r = signature [ 'r' ] . native
    s = signature [ 's' ] . native
except ( ValueError ) :
    raise SignatureError ( 'Signature is invalid' )
invalid = 0
# Check r is valid
invalid |= r < 1
invalid |= r >= n
# Check s is valid
invalid |= s < 1
invalid |= s >= n
if invalid :
    raise SignatureError ( 'Signature is invalid' )
hash_func = getattr ( hashlib , hash_algorithm )
digest = hash_func ( data ) . digest ( )
z = int_from_bytes ( digest , signed = False ) % n
w = inverse_mod ( s , n )
u1 = ( z * w ) % n
u2 = ( r * w ) % n
hash_point = ( curve_base_point * u1 ) + ( public_key_point * u2 )
if r != ( hash_point . x % n ) :
    raise SignatureError ( 'Signature is invalid' )


signature:
def system_path ( ) :
    """Tries to find a CA certs bundle in common locations
 : raises :
 OSError - when no valid CA certs bundle was found on the filesystem
 : return :
 The full filesystem path to a CA certs bundle file"""

implementation:
ca_path = None
# Common CA cert paths
paths = [ '/usr/lib/ssl/certs/ca-certificates.crt' , '/etc/ssl/certs/ca-certificates.crt' , '/etc/ssl/certs/ca-bundle.crt' , '/etc/pki/tls/certs/ca-bundle.crt' , '/etc/ssl/ca-bundle.pem' , '/usr/local/share/certs/ca-root-nss.crt' , '/etc/ssl/cert.pem' ]
# First try SSL _ CERT _ FILE
if 'SSL_CERT_FILE' in os . environ :
    paths . insert ( 0 , os . environ [ 'SSL_CERT_FILE' ] )
for path in paths :
    if os . path . exists ( path ) and os . path . getsize ( path ) > 0 :
        ca_path = path
        break
if not ca_path :
    raise OSError ( pretty_message ( '''
            Unable to find a CA certs bundle in common locations - try
            setting the SSL_CERT_FILE environmental variable
            ''' ) )
return ca_path


signature:
def extract_from_system ( cert_callback = None , callback_only_on_failure = False ) :
    """Extracts trusted CA certs from the system CA cert bundle
 : param cert _ callback :
 A callback that is called once for each certificate in the trust store .
 It should accept two parameters : an asn1crypto . x509 . Certificate object ,
 and a reason . The reason will be None if the certificate is being
 exported , otherwise it will be a unicode string of the reason it won ' t .
 : param callback _ only _ on _ failure :
 A boolean - if the callback should only be called when a certificate is
 not exported .
 : return :
 A list of 3 - element tuples :
 - 0 : a byte string of a DER - encoded certificate
 - 1 : a set of unicode strings that are OIDs of purposes to trust the
 certificate for
 - 2 : a set of unicode strings that are OIDs of purposes to reject the
 certificate for"""

implementation:
all_purposes = '2.5.29.37.0'
ca_path = system_path ( )
output = [ ]
with open ( ca_path , 'rb' ) as f :
    for armor_type , _ , cert_bytes in unarmor ( f . read ( ) , multiple = True ) : # Without more info , a certificate is trusted for all purposes
        if armor_type == 'CERTIFICATE' :
            if cert_callback :
                cert_callback ( Certificate . load ( cert_bytes ) , None )
            output . append ( ( cert_bytes , set ( ) , set ( ) ) )
        # The OpenSSL TRUSTED CERTIFICATE construct adds OIDs for trusted
        # and rejected purposes , so we extract that info .
        elif armor_type == 'TRUSTED CERTIFICATE' :
            cert , aux = TrustedCertificate . load ( cert_bytes )
            reject_all = False
            trust_oids = set ( )
            reject_oids = set ( )
            for purpose in aux [ 'trust' ] :
                if purpose . dotted == all_purposes :
                    trust_oids = set ( [ purpose . dotted ] )
                    break
                trust_oids . add ( purpose . dotted )
            for purpose in aux [ 'reject' ] :
                if purpose . dotted == all_purposes :
                    reject_all = True
                    break
                reject_oids . add ( purpose . dotted )
            if reject_all :
                if cert_callback :
                    cert_callback ( cert , 'explicitly distrusted' )
                continue
            if cert_callback and not callback_only_on_failure :
                cert_callback ( cert , None )
            output . append ( ( cert . dump ( ) , trust_oids , reject_oids ) )
return output


signature:
def extract_from_system ( cert_callback = None , callback_only_on_failure = False ) :
    """Extracts trusted CA certificates from the Windows certificate store
 : param cert _ callback :
 A callback that is called once for each certificate in the trust store .
 It should accept two parameters : an asn1crypto . x509 . Certificate object ,
 and a reason . The reason will be None if the certificate is being
 exported , otherwise it will be a unicode string of the reason it won ' t .
 : param callback _ only _ on _ failure :
 A boolean - if the callback should only be called when a certificate is
 not exported .
 : raises :
 OSError - when an error is returned by the OS crypto library
 : return :
 A list of 3 - element tuples :
 - 0 : a byte string of a DER - encoded certificate
 - 1 : a set of unicode strings that are OIDs of purposes to trust the
 certificate for
 - 2 : a set of unicode strings that are OIDs of purposes to reject the
 certificate for"""

implementation:
certificates = { }
processed = { }
now = datetime . datetime . utcnow ( )
for store in [ "ROOT" , "CA" ] :
    store_handle = crypt32 . CertOpenSystemStoreW ( null ( ) , store )
    handle_error ( store_handle )
    context_pointer = null ( )
    while True :
        context_pointer = crypt32 . CertEnumCertificatesInStore ( store_handle , context_pointer )
        if is_null ( context_pointer ) :
            break
        context = unwrap ( context_pointer )
        trust_all = False
        data = None
        digest = None
        if context . dwCertEncodingType != Crypt32Const . X509_ASN_ENCODING :
            continue
        data = bytes_from_buffer ( context . pbCertEncoded , int ( context . cbCertEncoded ) )
        digest = hashlib . sha1 ( data ) . digest ( )
        if digest in processed :
            continue
        processed [ digest ] = True
        cert_info = unwrap ( context . pCertInfo )
        not_before_seconds = _convert_filetime_to_timestamp ( cert_info . NotBefore )
        try :
            not_before = datetime . datetime . fromtimestamp ( not_before_seconds )
            if not_before > now :
                if cert_callback :
                    cert_callback ( Certificate . load ( data ) , 'not yet valid' )
                continue
        except ( ValueError , OSError ) : # If there is an error converting the not before timestamp ,
        # it is almost certainly because it is from too long ago ,
        # which means the cert is definitely valid by now .
            pass
        not_after_seconds = _convert_filetime_to_timestamp ( cert_info . NotAfter )
        try :
            not_after = datetime . datetime . fromtimestamp ( not_after_seconds )
            if not_after < now :
                if cert_callback :
                    cert_callback ( Certificate . load ( data ) , 'no longer valid' )
                continue
        except ( ValueError , OSError ) as e : # The only reason we would get an exception here is if the
        # expiration time is so far in the future that it can ' t be
        # used as a timestamp , or it is before 0 . If it is very far
        # in the future , the cert is still valid , so we only raise
        # an exception if the timestamp is less than zero .
            if not_after_seconds < 0 :
                message = e . args [ 0 ] + ' - ' + str_cls ( not_after_seconds )
                e . args = ( message , ) + e . args [ 1 : ]
                raise e
        trust_oids = set ( )
        reject_oids = set ( )
        # Here we grab the extended key usage properties that Windows
        # layers on top of the extended key usage extension that is
        # part of the certificate itself . For highest security , users
        # should only use certificates for the intersection of the two
        # lists of purposes . However , many seen to treat the OS trust
        # list as an override .
        to_read = new ( crypt32 , 'DWORD *' , 0 )
        res = crypt32 . CertGetEnhancedKeyUsage ( context_pointer , Crypt32Const . CERT_FIND_PROP_ONLY_ENHKEY_USAGE_FLAG , null ( ) , to_read )
        # Per the Microsoft documentation , if CRYPT _ E _ NOT _ FOUND is returned
        # from get _ error ( ) , it means the certificate is valid for all purposes
        error_code , _ = get_error ( )
        if not res and error_code != Crypt32Const . CRYPT_E_NOT_FOUND :
            handle_error ( res )
        if error_code == Crypt32Const . CRYPT_E_NOT_FOUND :
            trust_all = True
        else :
            usage_buffer = buffer_from_bytes ( deref ( to_read ) )
            res = crypt32 . CertGetEnhancedKeyUsage ( context_pointer , Crypt32Const . CERT_FIND_PROP_ONLY_ENHKEY_USAGE_FLAG , cast ( crypt32 , 'CERT_ENHKEY_USAGE *' , usage_buffer ) , to_read )
            handle_error ( res )
            key_usage_pointer = struct_from_buffer ( crypt32 , 'CERT_ENHKEY_USAGE' , usage_buffer )
            key_usage = unwrap ( key_usage_pointer )
            # Having no enhanced usage properties means a cert is distrusted
            if key_usage . cUsageIdentifier == 0 :
                if cert_callback :
                    cert_callback ( Certificate . load ( data ) , 'explicitly distrusted' )
                continue
            oids = array_from_pointer ( crypt32 , 'LPCSTR' , key_usage . rgpszUsageIdentifier , key_usage . cUsageIdentifier )
            for oid in oids :
                trust_oids . add ( oid . decode ( 'ascii' ) )
        cert = None
        # If the certificate is not under blanket trust , we have to
        # determine what purposes it is rejected for by diffing the
        # set of OIDs from the certificate with the OIDs that are
        # trusted .
        if not trust_all :
            cert = Certificate . load ( data )
            if cert . extended_key_usage_value :
                for cert_oid in cert . extended_key_usage_value :
                    oid = cert_oid . dotted
                    if oid not in trust_oids :
                        reject_oids . add ( oid )
        if cert_callback and not callback_only_on_failure :
            if cert is None :
                cert = Certificate . load ( data )
            cert_callback ( cert , None )
        certificates [ digest ] = ( data , trust_oids , reject_oids )
    result = crypt32 . CertCloseStore ( store_handle , 0 )
    handle_error ( result )
    store_handle = None
return certificates . values ( )


signature:
def _convert_filetime_to_timestamp ( filetime ) :
    """Windows returns times as 64 - bit unsigned longs that are the number
 of hundreds of nanoseconds since Jan 1 1601 . This converts it to
 a datetime object .
 : param filetime :
 A FILETIME struct object
 : return :
 An integer unix timestamp"""

implementation:
hundreds_nano_seconds = struct . unpack ( b'>Q' , struct . pack ( b'>LL' , filetime . dwHighDateTime , filetime . dwLowDateTime ) ) [ 0 ]
seconds_since_1601 = hundreds_nano_seconds / 10000000
return seconds_since_1601 - 11644473600


signature:
def extract_chain ( server_handshake_bytes ) :
    """Extracts the X . 509 certificates from the server handshake bytes for use
 when debugging
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : return :
 A list of asn1crypto . x509 . Certificate objects"""

implementation:
output = [ ]
chain_bytes = None
for record_type , _ , record_data in parse_tls_records ( server_handshake_bytes ) :
    if record_type != b'\x16' :
        continue
    for message_type , message_data in parse_handshake_messages ( record_data ) :
        if message_type == b'\x0b' :
            chain_bytes = message_data
            break
    if chain_bytes :
        break
if chain_bytes : # The first 3 bytes are the cert chain length
    pointer = 3
    while pointer < len ( chain_bytes ) :
        cert_length = int_from_bytes ( chain_bytes [ pointer : pointer + 3 ] )
        cert_start = pointer + 3
        cert_end = cert_start + cert_length
        pointer = cert_end
        cert_bytes = chain_bytes [ cert_start : cert_end ]
        output . append ( Certificate . load ( cert_bytes ) )
return output


signature:
def detect_client_auth_request ( server_handshake_bytes ) :
    """Determines if a CertificateRequest message is sent from the server asking
 the client for a certificate
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : return :
 A boolean - if a client certificate request was found"""

implementation:
for record_type , _ , record_data in parse_tls_records ( server_handshake_bytes ) :
    if record_type != b'\x16' :
        continue
    for message_type , message_data in parse_handshake_messages ( record_data ) :
        if message_type == b'\x0d' :
            return True
return False


signature:
def get_dh_params_length ( server_handshake_bytes ) :
    """Determines the length of the DH params from the ServerKeyExchange
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : return :
 None or an integer of the bit size of the DH parameters"""

implementation:
output = None
dh_params_bytes = None
for record_type , _ , record_data in parse_tls_records ( server_handshake_bytes ) :
    if record_type != b'\x16' :
        continue
    for message_type , message_data in parse_handshake_messages ( record_data ) :
        if message_type == b'\x0c' :
            dh_params_bytes = message_data
            break
    if dh_params_bytes :
        break
if dh_params_bytes :
    output = int_from_bytes ( dh_params_bytes [ 0 : 2 ] ) * 8
return output


signature:
def parse_alert ( server_handshake_bytes ) :
    """Parses the handshake for protocol alerts
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : return :
 None or an 2 - element tuple of integers :
 0 : 1 ( warning ) or 2 ( fatal )
 1 : The alert description ( see https : / / tools . ietf . org / html / rfc5246 # section - 7.2)"""

implementation:
for record_type , _ , record_data in parse_tls_records ( server_handshake_bytes ) :
    if record_type != b'\x15' :
        continue
    if len ( record_data ) != 2 :
        return None
    return ( int_from_bytes ( record_data [ 0 : 1 ] ) , int_from_bytes ( record_data [ 1 : 2 ] ) )
return None


signature:
def parse_session_info ( server_handshake_bytes , client_handshake_bytes ) :
    """Parse the TLS handshake from the client to the server to extract information
 including the cipher suite selected , if compression is enabled , the
 session id and if a new or reused session ticket exists .
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : param client _ handshake _ bytes :
 A byte string of the handshake data sent to the server
 : return :
 A dict with the following keys :
 - " protocol " : unicode string
 - " cipher _ suite " : unicode string
 - " compression " : boolean
 - " session _ id " : " new " , " reused " or None
 - " session _ ticket : " new " , " reused " or None"""

implementation:
protocol = None
cipher_suite = None
compression = False
session_id = None
session_ticket = None
server_session_id = None
client_session_id = None
for record_type , _ , record_data in parse_tls_records ( server_handshake_bytes ) :
    if record_type != b'\x16' :
        continue
    for message_type , message_data in parse_handshake_messages ( record_data ) : # Ensure we are working with a ServerHello message
        if message_type != b'\x02' :
            continue
        protocol = { b'\x03\x00' : "SSLv3" , b'\x03\x01' : "TLSv1" , b'\x03\x02' : "TLSv1.1" , b'\x03\x03' : "TLSv1.2" , b'\x03\x04' : "TLSv1.3" , } [ message_data [ 0 : 2 ] ]
        session_id_length = int_from_bytes ( message_data [ 34 : 35 ] )
        if session_id_length > 0 :
            server_session_id = message_data [ 35 : 35 + session_id_length ]
        cipher_suite_start = 35 + session_id_length
        cipher_suite_bytes = message_data [ cipher_suite_start : cipher_suite_start + 2 ]
        cipher_suite = CIPHER_SUITE_MAP [ cipher_suite_bytes ]
        compression_start = cipher_suite_start + 2
        compression = message_data [ compression_start : compression_start + 1 ] != b'\x00'
        extensions_length_start = compression_start + 1
        extensions_data = message_data [ extensions_length_start : ]
        for extension_type , extension_data in _parse_hello_extensions ( extensions_data ) :
            if extension_type == 35 :
                session_ticket = "new"
                break
        break
for record_type , _ , record_data in parse_tls_records ( client_handshake_bytes ) :
    if record_type != b'\x16' :
        continue
    for message_type , message_data in parse_handshake_messages ( record_data ) : # Ensure we are working with a ClientHello message
        if message_type != b'\x01' :
            continue
        session_id_length = int_from_bytes ( message_data [ 34 : 35 ] )
        if session_id_length > 0 :
            client_session_id = message_data [ 35 : 35 + session_id_length ]
        cipher_suite_start = 35 + session_id_length
        cipher_suite_length = int_from_bytes ( message_data [ cipher_suite_start : cipher_suite_start + 2 ] )
        compression_start = cipher_suite_start + 2 + cipher_suite_length
        compression_length = int_from_bytes ( message_data [ compression_start : compression_start + 1 ] )
        # On subsequent requests , the session ticket will only be seen
        # in the ClientHello message
        if server_session_id is None and session_ticket is None :
            extensions_length_start = compression_start + 1 + compression_length
            extensions_data = message_data [ extensions_length_start : ]
            for extension_type , extension_data in _parse_hello_extensions ( extensions_data ) :
                if extension_type == 35 :
                    session_ticket = "reused"
                    break
        break
if server_session_id is not None :
    if client_session_id is None :
        session_id = "new"
    else :
        if client_session_id != server_session_id :
            session_id = "new"
        else :
            session_id = "reused"
return { "protocol" : protocol , "cipher_suite" : cipher_suite , "compression" : compression , "session_id" : session_id , "session_ticket" : session_ticket , }


signature:
def parse_tls_records ( data ) :
    """Creates a generator returning tuples of information about each record
 in a byte string of data from a TLS client or server . Stops as soon as it
 find a ChangeCipherSpec message since all data from then on is encrypted .
 : param data :
 A byte string of TLS records
 : return :
 A generator that yields 3 - element tuples :
 [0 ] Byte string of record type
 [1 ] Byte string of protocol version
 [2 ] Byte string of record data"""

implementation:
pointer = 0
data_len = len ( data )
while pointer < data_len : # Don ' t try to parse any more once the ChangeCipherSpec is found
    if data [ pointer : pointer + 1 ] == b'\x14' :
        break
    length = int_from_bytes ( data [ pointer + 3 : pointer + 5 ] )
    yield ( data [ pointer : pointer + 1 ] , data [ pointer + 1 : pointer + 3 ] , data [ pointer + 5 : pointer + 5 + length ] )
    pointer += 5 + length


signature:
def parse_handshake_messages ( data ) :
    """Creates a generator returning tuples of information about each message in
 a byte string of data from a TLS handshake record
 : param data :
 A byte string of a TLS handshake record data
 : return :
 A generator that yields 2 - element tuples :
 [0 ] Byte string of message type
 [1 ] Byte string of message data"""

implementation:
pointer = 0
data_len = len ( data )
while pointer < data_len :
    length = int_from_bytes ( data [ pointer + 1 : pointer + 4 ] )
    yield ( data [ pointer : pointer + 1 ] , data [ pointer + 4 : pointer + 4 + length ] )
    pointer += 4 + length


signature:
def _parse_hello_extensions ( data ) :
    """Creates a generator returning tuples of information about each extension
 from a byte string of extension data contained in a ServerHello ores
 ClientHello message
 : param data :
 A byte string of a extension data from a TLS ServerHello or ClientHello
 message
 : return :
 A generator that yields 2 - element tuples :
 [0 ] Byte string of extension type
 [1 ] Byte string of extension data"""

implementation:
if data == b'' :
    return
extentions_length = int_from_bytes ( data [ 0 : 2 ] )
extensions_start = 2
extensions_end = 2 + extentions_length
pointer = extensions_start
while pointer < extensions_end :
    extension_type = int_from_bytes ( data [ pointer : pointer + 2 ] )
    extension_length = int_from_bytes ( data [ pointer + 2 : pointer + 4 ] )
    yield ( extension_type , data [ pointer + 4 : pointer + 4 + extension_length ] )
    pointer += 4 + extension_length


signature:
def raise_hostname ( certificate , hostname ) :
    """Raises a TLSVerificationError due to a hostname mismatch
 : param certificate :
 An asn1crypto . x509 . Certificate object
 : raises :
 TLSVerificationError"""

implementation:
is_ip = re . match ( '^\\d+\\.\\d+\\.\\d+\\.\\d+$' , hostname ) or hostname . find ( ':' ) != - 1
if is_ip :
    hostname_type = 'IP address %s' % hostname
else :
    hostname_type = 'domain name %s' % hostname
message = 'Server certificate verification failed - %s does not match' % hostname_type
valid_ips = ', ' . join ( certificate . valid_ips )
valid_domains = ', ' . join ( certificate . valid_domains )
if valid_domains :
    message += ' valid domains: %s' % valid_domains
if valid_domains and valid_ips :
    message += ' or'
if valid_ips :
    message += ' valid IP addresses: %s' % valid_ips
raise TLSVerificationError ( message , certificate )


signature:
def raise_expired_not_yet_valid ( certificate ) :
    """Raises a TLSVerificationError due to certificate being expired , or not yet
 being valid
 : param certificate :
 An asn1crypto . x509 . Certificate object
 : raises :
 TLSVerificationError"""

implementation:
validity = certificate [ 'tbs_certificate' ] [ 'validity' ]
not_after = validity [ 'not_after' ] . native
not_before = validity [ 'not_before' ] . native
now = datetime . now ( timezone . utc )
if not_before > now :
    formatted_before = not_before . strftime ( '%Y-%m-%d %H:%M:%SZ' )
    message = 'Server certificate verification failed - certificate not valid until %s' % formatted_before
elif not_after < now :
    formatted_after = not_after . strftime ( '%Y-%m-%d %H:%M:%SZ' )
    message = 'Server certificate verification failed - certificate expired %s' % formatted_after
raise TLSVerificationError ( message , certificate )


signature:
def detect_other_protocol ( server_handshake_bytes ) :
    """Looks at the server handshake bytes to try and detect a different protocol
 : param server _ handshake _ bytes :
 A byte string of the handshake data received from the server
 : return :
 None , or a unicode string of " ftp " , " http " , " imap " , " pop3 " , " smtp " """

implementation:
if server_handshake_bytes [ 0 : 5 ] == b'HTTP/' :
    return 'HTTP'
if server_handshake_bytes [ 0 : 4 ] == b'220 ' :
    if re . match ( b'^[^\r\n]*ftp' , server_handshake_bytes , re . I ) :
        return 'FTP'
    else :
        return 'SMTP'
if server_handshake_bytes [ 0 : 4 ] == b'220-' :
    return 'FTP'
if server_handshake_bytes [ 0 : 4 ] == b'+OK ' :
    return 'POP3'
if server_handshake_bytes [ 0 : 4 ] == b'* OK' or server_handshake_bytes [ 0 : 9 ] == b'* PREAUTH' :
    return 'IMAP'
return None


signature:
def constant_compare ( a , b ) :
    """Compares two byte strings in constant time to see if they are equal
 : param a :
 The first byte string
 : param b :
 The second byte string
 : return :
 A boolean if the two byte strings are equal"""

implementation:
if not isinstance ( a , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            a must be a byte string, not %s
            ''' , type_name ( a ) ) )
if not isinstance ( b , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            b must be a byte string, not %s
            ''' , type_name ( b ) ) )
if len ( a ) != len ( b ) :
    return False
if sys . version_info < ( 3 , ) :
    a = [ ord ( char ) for char in a ]
    b = [ ord ( char ) for char in b ]
result = 0
for x , y in zip ( a , b ) :
    result |= x ^ y
return result == 0


signature:
def _try_decode ( byte_string ) :
    """Tries decoding a byte string from the OS into a unicode string
 : param byte _ string :
 A byte string
 : return :
 A unicode string"""

implementation:
try :
    return str_cls ( byte_string , _encoding )
# If the " correct " encoding did not work , try some defaults , and then just
# obliterate characters that we can ' t seen to decode properly
except ( UnicodeDecodeError ) :
    for encoding in _fallback_encodings :
        try :
            return str_cls ( byte_string , encoding , errors = 'strict' )
        except ( UnicodeDecodeError ) :
            pass
return str_cls ( byte_string , errors = 'replace' )


signature:
def _read_callback ( connection_id , data_buffer , data_length_pointer ) :
    """Callback called by Secure Transport to actually read the socket
 : param connection _ id :
 An integer identifing the connection
 : param data _ buffer :
 A char pointer FFI type to write the data to
 : param data _ length _ pointer :
 A size _ t pointer FFI type of the amount of data to read . Will be
 overwritten with the amount of data read on return .
 : return :
 An integer status code of the result - 0 for success"""

implementation:
self = None
try :
    self = _connection_refs . get ( connection_id )
    if not self :
        socket = _socket_refs . get ( connection_id )
    else :
        socket = self . _socket
    if not self and not socket :
        return 0
    bytes_requested = deref ( data_length_pointer )
    timeout = socket . gettimeout ( )
    error = None
    data = b''
    try :
        while len ( data ) < bytes_requested : # Python 2 on Travis CI seems to have issues with blocking on
        # recv ( ) for longer than the socket timeout value , so we select
            if timeout is not None and timeout > 0.0 :
                read_ready , _ , _ = select . select ( [ socket ] , [ ] , [ ] , timeout )
                if len ( read_ready ) == 0 :
                    raise socket_ . error ( errno . EAGAIN , 'timed out' )
            chunk = socket . recv ( bytes_requested - len ( data ) )
            data += chunk
            if chunk == b'' :
                if len ( data ) == 0 :
                    if timeout is None :
                        return SecurityConst . errSSLClosedNoNotify
                    return SecurityConst . errSSLClosedAbort
                break
    except ( socket_ . error ) as e :
        error = e . errno
    if error is not None and error != errno . EAGAIN :
        if error == errno . ECONNRESET or error == errno . EPIPE :
            return SecurityConst . errSSLClosedNoNotify
        return SecurityConst . errSSLClosedAbort
    if self and not self . _done_handshake : # SecureTransport doesn ' t bother to check if the TLS record header
    # is valid before asking to read more data , which can result in
    # connection hangs . Here we do basic checks to get around the issue .
        if len ( data ) >= 3 and len ( self . _server_hello ) == 0 : # Check to ensure it is an alert or handshake first
            valid_record_type = data [ 0 : 1 ] in set ( [ b'\x15' , b'\x16' ] )
            # Check if the protocol version is SSL 3.0 or TLS 1.0-1.3
            valid_protocol_version = data [ 1 : 3 ] in set ( [ b'\x03\x00' , b'\x03\x01' , b'\x03\x02' , b'\x03\x03' , b'\x03\x04' ] )
            if not valid_record_type or not valid_protocol_version :
                self . _server_hello += data + _read_remaining ( socket )
                return SecurityConst . errSSLProtocol
        self . _server_hello += data
    write_to_buffer ( data_buffer , data )
    pointer_set ( data_length_pointer , len ( data ) )
    if len ( data ) != bytes_requested :
        return SecurityConst . errSSLWouldBlock
    return 0
except ( KeyboardInterrupt ) as e :
    if self :
        self . _exception = e
    return SecurityConst . errSSLClosedAbort


signature:
def _read_remaining ( socket ) :
    """Reads everything available from the socket - used for debugging when there
 is a protocol error
 : param socket :
 The socket to read from
 : return :
 A byte string of the remaining data"""

implementation:
output = b''
old_timeout = socket . gettimeout ( )
try :
    socket . settimeout ( 0.0 )
    output += socket . recv ( 8192 )
except ( socket_ . error ) :
    pass
finally :
    socket . settimeout ( old_timeout )
return output


signature:
def _write_callback ( connection_id , data_buffer , data_length_pointer ) :
    """Callback called by Secure Transport to actually write to the socket
 : param connection _ id :
 An integer identifing the connection
 : param data _ buffer :
 A char pointer FFI type containing the data to write
 : param data _ length _ pointer :
 A size _ t pointer FFI type of the amount of data to write . Will be
 overwritten with the amount of data actually written on return .
 : return :
 An integer status code of the result - 0 for success"""

implementation:
try :
    self = _connection_refs . get ( connection_id )
    if not self :
        socket = _socket_refs . get ( connection_id )
    else :
        socket = self . _socket
    if not self and not socket :
        return 0
    data_length = deref ( data_length_pointer )
    data = bytes_from_buffer ( data_buffer , data_length )
    if self and not self . _done_handshake :
        self . _client_hello += data
    error = None
    try :
        sent = socket . send ( data )
    except ( socket_ . error ) as e :
        error = e . errno
    if error is not None and error != errno . EAGAIN :
        if error == errno . ECONNRESET or error == errno . EPIPE :
            return SecurityConst . errSSLClosedNoNotify
        return SecurityConst . errSSLClosedAbort
    if sent != data_length :
        pointer_set ( data_length_pointer , sent )
        return SecurityConst . errSSLWouldBlock
    return 0
except ( KeyboardInterrupt ) as e :
    self . _exception = e
    return SecurityConst . errSSLPeerUserCancelled


signature:
def _handshake ( self ) :
    """Perform an initial TLS handshake"""

implementation:
session_context = None
ssl_policy_ref = None
crl_search_ref = None
crl_policy_ref = None
ocsp_search_ref = None
ocsp_policy_ref = None
policy_array_ref = None
try :
    if osx_version_info < ( 10 , 8 ) :
        session_context_pointer = new ( Security , 'SSLContextRef *' )
        result = Security . SSLNewContext ( False , session_context_pointer )
        handle_sec_error ( result )
        session_context = unwrap ( session_context_pointer )
    else :
        session_context = Security . SSLCreateContext ( null ( ) , SecurityConst . kSSLClientSide , SecurityConst . kSSLStreamType )
    result = Security . SSLSetIOFuncs ( session_context , _read_callback_pointer , _write_callback_pointer )
    handle_sec_error ( result )
    self . _connection_id = id ( self ) % 2147483647
    _connection_refs [ self . _connection_id ] = self
    _socket_refs [ self . _connection_id ] = self . _socket
    result = Security . SSLSetConnection ( session_context , self . _connection_id )
    handle_sec_error ( result )
    utf8_domain = self . _hostname . encode ( 'utf-8' )
    result = Security . SSLSetPeerDomainName ( session_context , utf8_domain , len ( utf8_domain ) )
    handle_sec_error ( result )
    if osx_version_info >= ( 10 , 10 ) :
        disable_auto_validation = self . _session . _manual_validation or self . _session . _extra_trust_roots
        explicit_validation = ( not self . _session . _manual_validation ) and self . _session . _extra_trust_roots
    else :
        disable_auto_validation = True
        explicit_validation = not self . _session . _manual_validation
    # Ensure requested protocol support is set for the session
    if osx_version_info < ( 10 , 8 ) :
        for protocol in [ 'SSLv2' , 'SSLv3' , 'TLSv1' ] :
            protocol_const = _PROTOCOL_STRING_CONST_MAP [ protocol ]
            enabled = protocol in self . _session . _protocols
            result = Security . SSLSetProtocolVersionEnabled ( session_context , protocol_const , enabled )
            handle_sec_error ( result )
        if disable_auto_validation :
            result = Security . SSLSetEnableCertVerify ( session_context , False )
            handle_sec_error ( result )
    else :
        protocol_consts = [ _PROTOCOL_STRING_CONST_MAP [ protocol ] for protocol in self . _session . _protocols ]
        min_protocol = min ( protocol_consts )
        max_protocol = max ( protocol_consts )
        result = Security . SSLSetProtocolVersionMin ( session_context , min_protocol )
        handle_sec_error ( result )
        result = Security . SSLSetProtocolVersionMax ( session_context , max_protocol )
        handle_sec_error ( result )
        if disable_auto_validation :
            result = Security . SSLSetSessionOption ( session_context , SecurityConst . kSSLSessionOptionBreakOnServerAuth , True )
            handle_sec_error ( result )
    # Disable all sorts of bad cipher suites
    supported_ciphers_pointer = new ( Security , 'size_t *' )
    result = Security . SSLGetNumberSupportedCiphers ( session_context , supported_ciphers_pointer )
    handle_sec_error ( result )
    supported_ciphers = deref ( supported_ciphers_pointer )
    cipher_buffer = buffer_from_bytes ( supported_ciphers * 4 )
    supported_cipher_suites_pointer = cast ( Security , 'uint32_t *' , cipher_buffer )
    result = Security . SSLGetSupportedCiphers ( session_context , supported_cipher_suites_pointer , supported_ciphers_pointer )
    handle_sec_error ( result )
    supported_ciphers = deref ( supported_ciphers_pointer )
    supported_cipher_suites = array_from_pointer ( Security , 'uint32_t' , supported_cipher_suites_pointer , supported_ciphers )
    good_ciphers = [ ]
    for supported_cipher_suite in supported_cipher_suites :
        cipher_suite = int_to_bytes ( supported_cipher_suite , width = 2 )
        cipher_suite_name = CIPHER_SUITE_MAP . get ( cipher_suite , cipher_suite )
        good_cipher = _cipher_blacklist_regex . search ( cipher_suite_name ) is None
        if good_cipher :
            good_ciphers . append ( supported_cipher_suite )
    num_good_ciphers = len ( good_ciphers )
    good_ciphers_array = new ( Security , 'uint32_t[]' , num_good_ciphers )
    array_set ( good_ciphers_array , good_ciphers )
    good_ciphers_pointer = cast ( Security , 'uint32_t *' , good_ciphers_array )
    result = Security . SSLSetEnabledCiphers ( session_context , good_ciphers_pointer , num_good_ciphers )
    handle_sec_error ( result )
    # Set a peer id from the session to allow for session reuse , the hostname
    # is appended to prevent a bug on OS X 10.7 where it tries to reuse a
    # connection even if the hostnames are different .
    peer_id = self . _session . _peer_id + self . _hostname . encode ( 'utf-8' )
    result = Security . SSLSetPeerID ( session_context , peer_id , len ( peer_id ) )
    handle_sec_error ( result )
    handshake_result = Security . SSLHandshake ( session_context )
    if self . _exception is not None :
        exception = self . _exception
        self . _exception = None
        raise exception
    while handshake_result == SecurityConst . errSSLWouldBlock :
        handshake_result = Security . SSLHandshake ( session_context )
        if self . _exception is not None :
            exception = self . _exception
            self . _exception = None
            raise exception
    if osx_version_info < ( 10 , 8 ) and osx_version_info >= ( 10 , 7 ) :
        do_validation = explicit_validation and handshake_result == 0
    else :
        do_validation = explicit_validation and handshake_result == SecurityConst . errSSLServerAuthCompleted
    if do_validation :
        trust_ref_pointer = new ( Security , 'SecTrustRef *' )
        result = Security . SSLCopyPeerTrust ( session_context , trust_ref_pointer )
        handle_sec_error ( result )
        trust_ref = unwrap ( trust_ref_pointer )
        cf_string_hostname = CFHelpers . cf_string_from_unicode ( self . _hostname )
        ssl_policy_ref = Security . SecPolicyCreateSSL ( True , cf_string_hostname )
        result = CoreFoundation . CFRelease ( cf_string_hostname )
        handle_cf_error ( result )
        # Create a new policy for OCSP checking to disable it
        ocsp_oid_pointer = struct ( Security , 'CSSM_OID' )
        ocsp_oid = unwrap ( ocsp_oid_pointer )
        ocsp_oid . Length = len ( SecurityConst . APPLE_TP_REVOCATION_OCSP )
        ocsp_oid_buffer = buffer_from_bytes ( SecurityConst . APPLE_TP_REVOCATION_OCSP )
        ocsp_oid . Data = cast ( Security , 'char *' , ocsp_oid_buffer )
        ocsp_search_ref_pointer = new ( Security , 'SecPolicySearchRef *' )
        result = Security . SecPolicySearchCreate ( SecurityConst . CSSM_CERT_X_509v3 , ocsp_oid_pointer , null ( ) , ocsp_search_ref_pointer )
        handle_sec_error ( result )
        ocsp_search_ref = unwrap ( ocsp_search_ref_pointer )
        ocsp_policy_ref_pointer = new ( Security , 'SecPolicyRef *' )
        result = Security . SecPolicySearchCopyNext ( ocsp_search_ref , ocsp_policy_ref_pointer )
        handle_sec_error ( result )
        ocsp_policy_ref = unwrap ( ocsp_policy_ref_pointer )
        ocsp_struct_pointer = struct ( Security , 'CSSM_APPLE_TP_OCSP_OPTIONS' )
        ocsp_struct = unwrap ( ocsp_struct_pointer )
        ocsp_struct . Version = SecurityConst . CSSM_APPLE_TP_OCSP_OPTS_VERSION
        ocsp_struct . Flags = ( SecurityConst . CSSM_TP_ACTION_OCSP_DISABLE_NET | SecurityConst . CSSM_TP_ACTION_OCSP_CACHE_READ_DISABLE )
        ocsp_struct_bytes = struct_bytes ( ocsp_struct_pointer )
        cssm_data_pointer = struct ( Security , 'CSSM_DATA' )
        cssm_data = unwrap ( cssm_data_pointer )
        cssm_data . Length = len ( ocsp_struct_bytes )
        ocsp_struct_buffer = buffer_from_bytes ( ocsp_struct_bytes )
        cssm_data . Data = cast ( Security , 'char *' , ocsp_struct_buffer )
        result = Security . SecPolicySetValue ( ocsp_policy_ref , cssm_data_pointer )
        handle_sec_error ( result )
        # Create a new policy for CRL checking to disable it
        crl_oid_pointer = struct ( Security , 'CSSM_OID' )
        crl_oid = unwrap ( crl_oid_pointer )
        crl_oid . Length = len ( SecurityConst . APPLE_TP_REVOCATION_CRL )
        crl_oid_buffer = buffer_from_bytes ( SecurityConst . APPLE_TP_REVOCATION_CRL )
        crl_oid . Data = cast ( Security , 'char *' , crl_oid_buffer )
        crl_search_ref_pointer = new ( Security , 'SecPolicySearchRef *' )
        result = Security . SecPolicySearchCreate ( SecurityConst . CSSM_CERT_X_509v3 , crl_oid_pointer , null ( ) , crl_search_ref_pointer )
        handle_sec_error ( result )
        crl_search_ref = unwrap ( crl_search_ref_pointer )
        crl_policy_ref_pointer = new ( Security , 'SecPolicyRef *' )
        result = Security . SecPolicySearchCopyNext ( crl_search_ref , crl_policy_ref_pointer )
        handle_sec_error ( result )
        crl_policy_ref = unwrap ( crl_policy_ref_pointer )
        crl_struct_pointer = struct ( Security , 'CSSM_APPLE_TP_CRL_OPTIONS' )
        crl_struct = unwrap ( crl_struct_pointer )
        crl_struct . Version = SecurityConst . CSSM_APPLE_TP_CRL_OPTS_VERSION
        crl_struct . CrlFlags = 0
        crl_struct_bytes = struct_bytes ( crl_struct_pointer )
        cssm_data_pointer = struct ( Security , 'CSSM_DATA' )
        cssm_data = unwrap ( cssm_data_pointer )
        cssm_data . Length = len ( crl_struct_bytes )
        crl_struct_buffer = buffer_from_bytes ( crl_struct_bytes )
        cssm_data . Data = cast ( Security , 'char *' , crl_struct_buffer )
        result = Security . SecPolicySetValue ( crl_policy_ref , cssm_data_pointer )
        handle_sec_error ( result )
        policy_array_ref = CFHelpers . cf_array_from_list ( [ ssl_policy_ref , crl_policy_ref , ocsp_policy_ref ] )
        result = Security . SecTrustSetPolicies ( trust_ref , policy_array_ref )
        handle_sec_error ( result )
        if self . _session . _extra_trust_roots :
            ca_cert_refs = [ ]
            ca_certs = [ ]
            for cert in self . _session . _extra_trust_roots :
                ca_cert = load_certificate ( cert )
                ca_certs . append ( ca_cert )
                ca_cert_refs . append ( ca_cert . sec_certificate_ref )
            result = Security . SecTrustSetAnchorCertificatesOnly ( trust_ref , False )
            handle_sec_error ( result )
            array_ref = CFHelpers . cf_array_from_list ( ca_cert_refs )
            result = Security . SecTrustSetAnchorCertificates ( trust_ref , array_ref )
            handle_sec_error ( result )
        result_pointer = new ( Security , 'SecTrustResultType *' )
        result = Security . SecTrustEvaluate ( trust_ref , result_pointer )
        handle_sec_error ( result )
        trust_result_code = deref ( result_pointer )
        invalid_chain_error_codes = set ( [ SecurityConst . kSecTrustResultProceed , SecurityConst . kSecTrustResultUnspecified ] )
        if trust_result_code not in invalid_chain_error_codes :
            handshake_result = SecurityConst . errSSLXCertChainInvalid
        else :
            handshake_result = Security . SSLHandshake ( session_context )
            while handshake_result == SecurityConst . errSSLWouldBlock :
                handshake_result = Security . SSLHandshake ( session_context )
    self . _done_handshake = True
    handshake_error_codes = set ( [ SecurityConst . errSSLXCertChainInvalid , SecurityConst . errSSLCertExpired , SecurityConst . errSSLCertNotYetValid , SecurityConst . errSSLUnknownRootCert , SecurityConst . errSSLNoRootCert , SecurityConst . errSSLHostNameMismatch , SecurityConst . errSSLInternal , ] )
    # In testing , only errSSLXCertChainInvalid was ever returned for
    # all of these different situations , however we include the others
    # for completeness . To get the real reason we have to use the
    # certificate from the handshake and use the deprecated function
    # SecTrustGetCssmResultCode ( ) .
    if handshake_result in handshake_error_codes :
        trust_ref_pointer = new ( Security , 'SecTrustRef *' )
        result = Security . SSLCopyPeerTrust ( session_context , trust_ref_pointer )
        handle_sec_error ( result )
        trust_ref = unwrap ( trust_ref_pointer )
        result_code_pointer = new ( Security , 'OSStatus *' )
        result = Security . SecTrustGetCssmResultCode ( trust_ref , result_code_pointer )
        result_code = deref ( result_code_pointer )
        chain = extract_chain ( self . _server_hello )
        self_signed = False
        revoked = False
        expired = False
        not_yet_valid = False
        no_issuer = False
        cert = None
        bad_hostname = False
        if chain :
            cert = chain [ 0 ]
            oscrypto_cert = load_certificate ( cert )
            self_signed = oscrypto_cert . self_signed
            revoked = result_code == SecurityConst . CSSMERR_TP_CERT_REVOKED
            no_issuer = not self_signed and result_code == SecurityConst . CSSMERR_TP_NOT_TRUSTED
            expired = result_code == SecurityConst . CSSMERR_TP_CERT_EXPIRED
            not_yet_valid = result_code == SecurityConst . CSSMERR_TP_CERT_NOT_VALID_YET
            bad_hostname = result_code == SecurityConst . CSSMERR_APPLETP_HOSTNAME_MISMATCH
            # On macOS 10.12 , some expired certificates return errSSLInternal
            if osx_version_info >= ( 10 , 12 ) :
                validity = cert [ 'tbs_certificate' ] [ 'validity' ]
                not_before = validity [ 'not_before' ] . chosen . native
                not_after = validity [ 'not_after' ] . chosen . native
                utcnow = datetime . datetime . now ( timezone . utc )
                expired = not_after < utcnow
                not_yet_valid = not_before > utcnow
        if chain and chain [ 0 ] . hash_algo in set ( [ 'md5' , 'md2' ] ) :
            raise_weak_signature ( chain [ 0 ] )
        if revoked :
            raise_revoked ( cert )
        if bad_hostname :
            raise_hostname ( cert , self . _hostname )
        elif expired or not_yet_valid :
            raise_expired_not_yet_valid ( cert )
        elif no_issuer :
            raise_no_issuer ( cert )
        elif self_signed :
            raise_self_signed ( cert )
        if detect_client_auth_request ( self . _server_hello ) :
            raise_client_auth ( )
        raise_verification ( cert )
    if handshake_result == SecurityConst . errSSLPeerHandshakeFail :
        if detect_client_auth_request ( self . _server_hello ) :
            raise_client_auth ( )
        raise_handshake ( )
    if handshake_result == SecurityConst . errSSLWeakPeerEphemeralDHKey :
        raise_dh_params ( )
    if handshake_result == SecurityConst . errSSLPeerProtocolVersion :
        raise_protocol_version ( )
    if handshake_result in set ( [ SecurityConst . errSSLRecordOverflow , SecurityConst . errSSLProtocol ] ) :
        self . _server_hello += _read_remaining ( self . _socket )
        raise_protocol_error ( self . _server_hello )
    if handshake_result in set ( [ SecurityConst . errSSLClosedNoNotify , SecurityConst . errSSLClosedAbort ] ) :
        if not self . _done_handshake :
            self . _server_hello += _read_remaining ( self . _socket )
        if detect_other_protocol ( self . _server_hello ) :
            raise_protocol_error ( self . _server_hello )
        raise_disconnection ( )
    if osx_version_info < ( 10 , 10 ) :
        dh_params_length = get_dh_params_length ( self . _server_hello )
        if dh_params_length is not None and dh_params_length < 1024 :
            raise_dh_params ( )
    would_block = handshake_result == SecurityConst . errSSLWouldBlock
    server_auth_complete = handshake_result == SecurityConst . errSSLServerAuthCompleted
    manual_validation = self . _session . _manual_validation and server_auth_complete
    if not would_block and not manual_validation :
        handle_sec_error ( handshake_result , TLSError )
    self . _session_context = session_context
    protocol_const_pointer = new ( Security , 'SSLProtocol *' )
    result = Security . SSLGetNegotiatedProtocolVersion ( session_context , protocol_const_pointer )
    handle_sec_error ( result )
    protocol_const = deref ( protocol_const_pointer )
    self . _protocol = _PROTOCOL_CONST_STRING_MAP [ protocol_const ]
    cipher_int_pointer = new ( Security , 'SSLCipherSuite *' )
    result = Security . SSLGetNegotiatedCipher ( session_context , cipher_int_pointer )
    handle_sec_error ( result )
    cipher_int = deref ( cipher_int_pointer )
    cipher_bytes = int_to_bytes ( cipher_int , width = 2 )
    self . _cipher_suite = CIPHER_SUITE_MAP . get ( cipher_bytes , cipher_bytes )
    session_info = parse_session_info ( self . _server_hello , self . _client_hello )
    self . _compression = session_info [ 'compression' ]
    self . _session_id = session_info [ 'session_id' ]
    self . _session_ticket = session_info [ 'session_ticket' ]
except ( OSError , socket_ . error ) :
    if session_context :
        if osx_version_info < ( 10 , 8 ) :
            result = Security . SSLDisposeContext ( session_context )
            handle_sec_error ( result )
        else :
            result = CoreFoundation . CFRelease ( session_context )
            handle_cf_error ( result )
    self . _session_context = None
    self . close ( )
    raise
finally : # Trying to release crl _ search _ ref or ocsp _ search _ ref results in
# a segmentation fault , so we do not do that
    if ssl_policy_ref :
        result = CoreFoundation . CFRelease ( ssl_policy_ref )
        handle_cf_error ( result )
        ssl_policy_ref = None
    if crl_policy_ref :
        result = CoreFoundation . CFRelease ( crl_policy_ref )
        handle_cf_error ( result )
        crl_policy_ref = None
    if ocsp_policy_ref :
        result = CoreFoundation . CFRelease ( ocsp_policy_ref )
        handle_cf_error ( result )
        ocsp_policy_ref = None
    if policy_array_ref :
        result = CoreFoundation . CFRelease ( policy_array_ref )
        handle_cf_error ( result )
        policy_array_ref = None


signature:
def read ( self , max_length ) :
    """Reads data from the TLS - wrapped socket
 : param max _ length :
 The number of bytes to read - output may be less than this
 : raises :
 socket . socket - when a non - TLS socket error occurs
 oscrypto . errors . TLSError - when a TLS - related error occurs
 oscrypto . errors . TLSDisconnectError - when the connection disconnects
 oscrypto . errors . TLSGracefulDisconnectError - when the remote end gracefully closed the connection
 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 . _session_context is None : # Even if the session is closed , we can use
# buffered data to respond to read requests
    if self . _decrypted_bytes != b'' :
        output = self . _decrypted_bytes
        self . _decrypted_bytes = b''
        return output
    self . _raise_closed ( )
buffered_length = len ( self . _decrypted_bytes )
# If we already have enough buffered data , just use that
if buffered_length >= max_length :
    output = self . _decrypted_bytes [ 0 : max_length ]
    self . _decrypted_bytes = self . _decrypted_bytes [ max_length : ]
    return output
# Don ' t block if we have buffered data available , since it is ok to
# return less than the max _ length
if buffered_length > 0 and not self . select_read ( 0 ) :
    output = self . _decrypted_bytes
    self . _decrypted_bytes = b''
    return output
# Only read enough to get the requested amount when
# combined with buffered data
to_read = max_length - len ( self . _decrypted_bytes )
read_buffer = buffer_from_bytes ( to_read )
processed_pointer = new ( Security , 'size_t *' )
result = Security . SSLRead ( self . _session_context , read_buffer , to_read , processed_pointer )
if self . _exception is not None :
    exception = self . _exception
    self . _exception = None
    raise exception
if result and result not in set ( [ SecurityConst . errSSLWouldBlock , SecurityConst . errSSLClosedGraceful ] ) :
    handle_sec_error ( result , TLSError )
if result and result == SecurityConst . errSSLClosedGraceful :
    self . _gracefully_closed = True
    self . _shutdown ( False )
    self . _raise_closed ( )
bytes_read = deref ( processed_pointer )
output = self . _decrypted_bytes + bytes_from_buffer ( read_buffer , bytes_read )
self . _decrypted_bytes = output [ max_length : ]
return output [ 0 : max_length ]


signature:
def read_until ( self , marker ) :
    """Reads data from the socket until a marker is found . Data read includes
 the marker .
 : param marker :
 A byte string or regex object from re . compile ( ) . Used to determine
 when to stop reading . Regex objects are more inefficient since
 they must scan the entire byte string of read data each time data
 is read off the socket .
 : return :
 A byte string of the data read , including the marker"""

implementation:
if not isinstance ( marker , byte_cls ) and not isinstance ( marker , Pattern ) :
    raise TypeError ( pretty_message ( '''
                marker must be a byte string or compiled regex object, not %s
                ''' , type_name ( marker ) ) )
output = b''
is_regex = isinstance ( marker , Pattern )
while True :
    if len ( self . _decrypted_bytes ) > 0 :
        chunk = self . _decrypted_bytes
        self . _decrypted_bytes = b''
    else :
        to_read = self . _os_buffered_size ( ) or 8192
        chunk = self . read ( to_read )
    offset = len ( output )
    output += chunk
    if is_regex :
        match = marker . search ( output )
        if match is not None :
            end = match . end ( )
            break
    else : # If the marker was not found last time , we have to start
    # at a position where the marker would have its final char
    # in the newly read chunk
        start = max ( 0 , offset - len ( marker ) - 1 )
        match = output . find ( marker , start )
        if match != - 1 :
            end = match + len ( marker )
            break
self . _decrypted_bytes = output [ end : ] + self . _decrypted_bytes
return output [ 0 : end ]


signature:
def _os_buffered_size ( self ) :
    """Returns the number of bytes of decrypted data stored in the Secure
 Transport read buffer . This amount of data can be read from SSLRead ( )
 without calling self . _ socket . recv ( ) .
 : return :
 An integer - the number of available bytes"""

implementation:
num_bytes_pointer = new ( Security , 'size_t *' )
result = Security . SSLGetBufferedReadSize ( self . _session_context , num_bytes_pointer )
handle_sec_error ( result )
return deref ( num_bytes_pointer )


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
 oscrypto . errors . TLSDisconnectError - when the connection disconnects
 oscrypto . errors . TLSGracefulDisconnectError - when the remote end gracefully closed the connection
 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 . _session_context is None :
    self . _raise_closed ( )
processed_pointer = new ( Security , 'size_t *' )
data_len = len ( data )
while data_len :
    write_buffer = buffer_from_bytes ( data )
    result = Security . SSLWrite ( self . _session_context , write_buffer , data_len , processed_pointer )
    if self . _exception is not None :
        exception = self . _exception
        self . _exception = None
        raise exception
    handle_sec_error ( result , TLSError )
    bytes_written = deref ( processed_pointer )
    data = data [ bytes_written : ]
    data_len = len ( data )
    if data_len > 0 :
        self . select_write ( )


signature:
def _shutdown ( self , manual ) :
    """Shuts down the TLS session and then shuts down the underlying socket
 : param manual :
 A boolean if the connection was manually shutdown"""

implementation:
if self . _session_context is None :
    return
# Ignore error during close in case other end closed already
result = Security . SSLClose ( self . _session_context )
if osx_version_info < ( 10 , 8 ) :
    result = Security . SSLDisposeContext ( self . _session_context )
    handle_sec_error ( result )
else :
    result = CoreFoundation . CFRelease ( self . _session_context )
    handle_cf_error ( result )
self . _session_context = None
if manual :
    self . _local_closed = True
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
    if self . _connection_id in _socket_refs :
        del _socket_refs [ self . _connection_id ]


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

implementation:
trust_ref = None
cf_data_ref = None
result = None
try :
    trust_ref_pointer = new ( Security , 'SecTrustRef *' )
    result = Security . SSLCopyPeerTrust ( self . _session_context , trust_ref_pointer )
    handle_sec_error ( result )
    trust_ref = unwrap ( trust_ref_pointer )
    number_certs = Security . SecTrustGetCertificateCount ( trust_ref )
    self . _intermediates = [ ]
    for index in range ( 0 , number_certs ) :
        sec_certificate_ref = Security . SecTrustGetCertificateAtIndex ( trust_ref , index )
        cf_data_ref = Security . SecCertificateCopyData ( sec_certificate_ref )
        cert_data = CFHelpers . cf_data_to_bytes ( cf_data_ref )
        result = CoreFoundation . CFRelease ( cf_data_ref )
        handle_cf_error ( result )
        cf_data_ref = None
        cert = x509 . Certificate . load ( cert_data )
        if index == 0 :
            self . _certificate = cert
        else :
            self . _intermediates . append ( cert )
finally :
    if trust_ref :
        result = CoreFoundation . CFRelease ( trust_ref )
        handle_cf_error ( result )
    if cf_data_ref :
        result = CoreFoundation . CFRelease ( cf_data_ref )
        handle_cf_error ( result )


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

implementation:
if self . _session_context 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 . _session_context is None :
    self . _raise_closed ( )
if self . _certificate is None :
    self . _read_certificates ( )
return self . _intermediates


signature:
def get_path ( temp_dir = None , cache_length = 24 , cert_callback = None ) :
    """Get the filesystem path to a file that contains OpenSSL - compatible CA certs .
 On OS X and Windows , there are extracted from the system certificate store
 and cached in a file on the filesystem . This path should not be writable
 by other users , otherwise they could inject CA certs into the trust list .
 : param temp _ dir :
 The temporary directory to cache the CA certs in on OS X and Windows .
 Needs to have secure permissions so other users can not modify the
 contents .
 : param cache _ length :
 The number of hours to cache the CA certs on OS X and Windows
 : param cert _ callback :
 A callback that is called once for each certificate in the trust store .
 It should accept two parameters : an asn1crypto . x509 . Certificate object ,
 and a reason . The reason will be None if the certificate is being
 exported , otherwise it will be a unicode string of the reason it won ' t .
 This is only called on Windows and OS X when passed to this function .
 : raises :
 oscrypto . errors . CACertsError - when an error occurs exporting / locating certs
 : return :
 The full filesystem path to a CA certs file"""

implementation:
ca_path , temp = _ca_path ( temp_dir )
# Windows and OS X
if temp and _cached_path_needs_update ( ca_path , cache_length ) :
    empty_set = set ( )
    any_purpose = '2.5.29.37.0'
    apple_ssl = '1.2.840.113635.100.1.3'
    win_server_auth = '1.3.6.1.5.5.7.3.1'
    with path_lock :
        if _cached_path_needs_update ( ca_path , cache_length ) :
            with open ( ca_path , 'wb' ) as f :
                for cert , trust_oids , reject_oids in extract_from_system ( cert_callback , True ) :
                    if sys . platform == 'darwin' :
                        if trust_oids != empty_set and any_purpose not in trust_oids and apple_ssl not in trust_oids :
                            if cert_callback :
                                cert_callback ( Certificate . load ( cert ) , 'implicitly distrusted for TLS' )
                            continue
                        if reject_oids != empty_set and ( apple_ssl in reject_oids or any_purpose in reject_oids ) :
                            if cert_callback :
                                cert_callback ( Certificate . load ( cert ) , 'explicitly distrusted for TLS' )
                            continue
                    elif sys . platform == 'win32' :
                        if trust_oids != empty_set and any_purpose not in trust_oids and win_server_auth not in trust_oids :
                            if cert_callback :
                                cert_callback ( Certificate . load ( cert ) , 'implicitly distrusted for TLS' )
                            continue
                        if reject_oids != empty_set and ( win_server_auth in reject_oids or any_purpose in reject_oids ) :
                            if cert_callback :
                                cert_callback ( Certificate . load ( cert ) , 'explicitly distrusted for TLS' )
                            continue
                    if cert_callback :
                        cert_callback ( Certificate . load ( cert ) , None )
                    f . write ( armor ( 'CERTIFICATE' , cert ) )
if not ca_path :
    raise CACertsError ( 'No CA certs found' )
return ca_path


signature:
def get_list ( cache_length = 24 , map_vendor_oids = True , cert_callback = None ) :
    """Retrieves ( and caches in memory ) the list of CA certs from the OS . Includes
 trust information from the OS - purposes the certificate should be trusted
 or rejected for .
 Trust information is encoded via object identifiers ( OIDs ) that are sourced
 from various RFCs and vendors ( Apple and Microsoft ) . This trust information
 augments what is in the certificate itself . Any OID that is in the set of
 trusted purposes indicates the certificate has been explicitly trusted for
 a purpose beyond the extended key purpose extension . Any OID in the reject
 set is a purpose that the certificate should not be trusted for , even if
 present in the extended key purpose extension .
 * A list of common trust OIDs can be found as part of the ` KeyPurposeId ( ) `
 class in the ` asn1crypto . x509 ` module of the ` asn1crypto ` package . *
 : param cache _ length :
 The number of hours to cache the CA certs in memory before they are
 refreshed
 : param map _ vendor _ oids :
 A bool indicating if the following mapping of OIDs should happen for
 trust information from the OS trust list :
 - 1.2.840.113635.100.1.3 ( apple _ ssl ) - > 1.3.6.1.5.5.7.3.1 ( server _ auth )
 - 1.2.840.113635.100.1.3 ( apple _ ssl ) - > 1.3.6.1.5.5.7.3.2 ( client _ auth )
 - 1.2.840.113635.100.1.8 ( apple _ smime ) - > 1.3.6.1.5.5.7.3.4 ( email _ protection )
 - 1.2.840.113635.100.1.9 ( apple _ eap ) - > 1.3.6.1.5.5.7.3.13 ( eap _ over _ ppp )
 - 1.2.840.113635.100.1.9 ( apple _ eap ) - > 1.3.6.1.5.5.7.3.14 ( eap _ over _ lan )
 - 1.2.840.113635.100.1.11 ( apple _ ipsec ) - > 1.3.6.1.5.5.7.3.5 ( ipsec _ end _ system )
 - 1.2.840.113635.100.1.11 ( apple _ ipsec ) - > 1.3.6.1.5.5.7.3.6 ( ipsec _ tunnel )
 - 1.2.840.113635.100.1.11 ( apple _ ipsec ) - > 1.3.6.1.5.5.7.3.7 ( ipsec _ user )
 - 1.2.840.113635.100.1.11 ( apple _ ipsec ) - > 1.3.6.1.5.5.7.3.17 ( ipsec _ ike )
 - 1.2.840.113635.100.1.16 ( apple _ code _ signing ) - > 1.3.6.1.5.5.7.3.3 ( code _ signing )
 - 1.2.840.113635.100.1.20 ( apple _ time _ stamping ) - > 1.3.6.1.5.5.7.3.8 ( time _ stamping )
 - 1.3.6.1.4.1.311.10.3.2 ( microsoft _ time _ stamp _ signing ) - > 1.3.6.1.5.5.7.3.8 ( time _ stamping )
 : param cert _ callback :
 A callback that is called once for each certificate in the trust store .
 It should accept two parameters : an asn1crypto . x509 . Certificate object ,
 and a reason . The reason will be None if the certificate is being
 exported , otherwise it will be a unicode string of the reason it won ' t .
 : raises :
 oscrypto . errors . CACertsError - when an error occurs exporting / locating certs
 : return :
 A ( copied ) list of 3 - element tuples containing CA certs from the OS
 trust ilst :
 - 0 : an asn1crypto . x509 . Certificate object
 - 1 : a set of unicode strings of OIDs of trusted purposes
 - 2 : a set of unicode strings of OIDs of rejected purposes"""

implementation:
if not _in_memory_up_to_date ( cache_length ) :
    with memory_lock :
        if not _in_memory_up_to_date ( cache_length ) :
            certs = [ ]
            for cert_bytes , trust_oids , reject_oids in extract_from_system ( cert_callback ) :
                if map_vendor_oids :
                    trust_oids = _map_oids ( trust_oids )
                    reject_oids = _map_oids ( reject_oids )
                certs . append ( ( Certificate . load ( cert_bytes ) , trust_oids , reject_oids ) )
            _module_values [ 'certs' ] = certs
            _module_values [ 'last_update' ] = time . time ( )
return list ( _module_values [ 'certs' ] )


signature:
def clear_cache ( temp_dir = None ) :
    """Clears any cached info that was exported from the OS trust store . This will
 ensure the latest changes are returned from calls to get _ list ( ) and
 get _ path ( ) , but at the expense of re - exporting and parsing all certificates .
 : param temp _ dir :
 The temporary directory to cache the CA certs in on OS X and Windows .
 Needs to have secure permissions so other users can not modify the
 contents . Must be the same value passed to get _ path ( ) ."""

implementation:
with memory_lock :
    _module_values [ 'last_update' ] = None
    _module_values [ 'certs' ] = None
ca_path , temp = _ca_path ( temp_dir )
if temp :
    with path_lock :
        if os . path . exists ( ca_path ) :
            os . remove ( ca_path )


signature:
def _ca_path ( temp_dir = None ) :
    """Returns the file path to the CA certs file
 : param temp _ dir :
 The temporary directory to cache the CA certs in on OS X and Windows .
 Needs to have secure permissions so other users can not modify the
 contents .
 : return :
 A 2 - element tuple :
 - 0 : A unicode string of the file path
 - 1 : A bool if the file is a temporary file"""

implementation:
ca_path = system_path ( )
# Windows and OS X
if ca_path is None :
    if temp_dir is None :
        temp_dir = tempfile . gettempdir ( )
    if not os . path . isdir ( temp_dir ) :
        raise CACertsError ( pretty_message ( '''
                The temp dir specified, "%s", is not a directory
                ''' , temp_dir ) )
    ca_path = os . path . join ( temp_dir , 'oscrypto-ca-bundle.crt' )
    return ( ca_path , True )
return ( ca_path , False )


implementation:
new_oids = set ( )
for oid in oids :
    if oid in _oid_map :
        new_oids |= _oid_map [ oid ]
return oids | new_oids


signature:
def _cached_path_needs_update ( ca_path , cache_length ) :
    """Checks to see if a cache file needs to be refreshed
 : param ca _ path :
 A unicode string of the path to the cache file
 : param cache _ length :
 An integer representing the number of hours the cache is valid for
 : return :
 A boolean - True if the cache needs to be updated , False if the file
 is up - to - date"""

implementation:
exists = os . path . exists ( ca_path )
if not exists :
    return True
stats = os . stat ( ca_path )
if stats . st_mtime < time . time ( ) - cache_length * 60 * 60 :
    return True
if stats . st_size == 0 :
    return True
return False


signature:
def rand_bytes ( length ) :
    """Returns a number of random bytes suitable for cryptographic purposes
 : param length :
 The desired number of bytes
 : 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 OpenSSL
 : return :
 A byte string"""

implementation:
if not isinstance ( length , int_types ) :
    raise TypeError ( pretty_message ( '''
            length must be an integer, not %s
            ''' , type_name ( length ) ) )
if length < 1 :
    raise ValueError ( 'length must be greater than 0' )
if length > 1024 :
    raise ValueError ( 'length must not be greater than 1024' )
return os . urandom ( length )


signature:
def pbkdf2 ( hash_algorithm , password , salt , iterations , key_length ) :
    """Implements PBKDF2 from PKCS # 5 v2.2 in pure Python
 : param hash _ algorithm :
 The string name of the hash algorithm to use : " md5 " , " sha1 " , " sha224 " ,
 " sha256 " , " sha384 " , " sha512"
 : param password :
 A byte string of the password to use an input to the KDF
 : param salt :
 A cryptographic random byte string
 : param iterations :
 The numbers of iterations to use when deriving the key
 : param key _ length :
 The length of the desired key in bytes
 : return :
 The derived key as a byte string"""

implementation:
if not isinstance ( password , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            password must be a byte string, not %s
            ''' , type_name ( password ) ) )
if not isinstance ( salt , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            salt must be a byte string, not %s
            ''' , type_name ( salt ) ) )
if not isinstance ( iterations , int_types ) :
    raise TypeError ( pretty_message ( '''
            iterations must be an integer, not %s
            ''' , type_name ( iterations ) ) )
if iterations < 1 :
    raise ValueError ( pretty_message ( '''
            iterations must be greater than 0 - is %s
            ''' , repr ( iterations ) ) )
if not isinstance ( key_length , int_types ) :
    raise TypeError ( pretty_message ( '''
            key_length must be an integer, not %s
            ''' , type_name ( key_length ) ) )
if key_length < 1 :
    raise ValueError ( pretty_message ( '''
            key_length must be greater than 0 - is %s
            ''' , repr ( key_length ) ) )
if hash_algorithm not in set ( [ 'md5' , 'sha1' , 'sha224' , 'sha256' , 'sha384' , 'sha512' ] ) :
    raise ValueError ( pretty_message ( '''
            hash_algorithm must be one of "md5", "sha1", "sha224", "sha256",
            "sha384", "sha512", not %s
            ''' , repr ( hash_algorithm ) ) )
algo = getattr ( hashlib , hash_algorithm )
hash_length = { 'md5' : 16 , 'sha1' : 20 , 'sha224' : 28 , 'sha256' : 32 , 'sha384' : 48 , 'sha512' : 64 } [ hash_algorithm ]
blocks = int ( math . ceil ( key_length / hash_length ) )
original_hmac = hmac . new ( password , None , algo )
int_pack = struct . Struct ( b'>I' ) . pack
output = b''
for block in range ( 1 , blocks + 1 ) :
    prf = original_hmac . copy ( )
    prf . update ( salt + int_pack ( block ) )
    last = prf . digest ( )
    u = int_from_bytes ( last )
    for _ in range ( 2 , iterations + 1 ) :
        prf = original_hmac . copy ( )
        prf . update ( last )
        last = prf . digest ( )
        u ^= int_from_bytes ( last )
    t = int_to_bytes ( u )
    output += t
return output [ 0 : key_length ]


signature:
def generate_pair ( algorithm , bit_size = None , curve = None ) :
    """Generates a public / private key pair
 : param algorithm :
 The key algorithm - " rsa " , " dsa " or " ec "
 : param bit _ size :
 An integer - used for " rsa " and " dsa " . For " rsa " the value maye be 1024,
 2048 , 3072 or 4096 . For " dsa " the value may be 1024.
 : param curve :
 A unicode string - used for " ec " keys . Valid values include " secp256r1 " ,
 " secp384r1 " and " secp521r1 " .
 : 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 2 - element tuple of ( PublicKey , PrivateKey ) . The contents of each key
 may be saved by calling . asn1 . dump ( ) ."""

implementation:
if algorithm not in set ( [ 'rsa' , 'dsa' , 'ec' ] ) :
    raise ValueError ( pretty_message ( '''
            algorithm must be one of "rsa", "dsa", "ec", not %s
            ''' , repr ( algorithm ) ) )
if algorithm == 'rsa' :
    if bit_size not in set ( [ 1024 , 2048 , 3072 , 4096 ] ) :
        raise ValueError ( pretty_message ( '''
                bit_size must be one of 1024, 2048, 3072, 4096, not %s
                ''' , repr ( bit_size ) ) )
elif algorithm == 'dsa' :
    if bit_size not in set ( [ 1024 ] ) :
        raise ValueError ( pretty_message ( '''
                bit_size must be 1024, not %s
                ''' , repr ( bit_size ) ) )
elif algorithm == 'ec' :
    if curve not in set ( [ 'secp256r1' , 'secp384r1' , 'secp521r1' ] ) :
        raise ValueError ( pretty_message ( '''
                curve must be one of "secp256r1", "secp384r1", "secp521r1", not %s
                ''' , repr ( curve ) ) )
cf_dict = None
public_key_ref = None
private_key_ref = None
cf_data_public = None
cf_data_private = None
cf_string = None
sec_access_ref = None
try :
    key_type = { 'dsa' : Security . kSecAttrKeyTypeDSA , 'ec' : Security . kSecAttrKeyTypeECDSA , 'rsa' : Security . kSecAttrKeyTypeRSA , } [ algorithm ]
    if algorithm == 'ec' :
        key_size = { 'secp256r1' : 256 , 'secp384r1' : 384 , 'secp521r1' : 521 , } [ curve ]
    else :
        key_size = bit_size
    private_key_pointer = new ( Security , 'SecKeyRef *' )
    public_key_pointer = new ( Security , 'SecKeyRef *' )
    cf_string = CFHelpers . cf_string_from_unicode ( "Temporary key from oscrypto python library - safe to delete" )
    # For some reason Apple decided that DSA keys were not a valid type of
    # key to be generated via SecKeyGeneratePair ( ) , thus we have to use the
    # lower level , deprecated SecKeyCreatePair ( )
    if algorithm == 'dsa' :
        sec_access_ref_pointer = new ( Security , 'SecAccessRef *' )
        result = Security . SecAccessCreate ( cf_string , null ( ) , sec_access_ref_pointer )
        sec_access_ref = unwrap ( sec_access_ref_pointer )
        result = Security . SecKeyCreatePair ( null ( ) , SecurityConst . CSSM_ALGID_DSA , key_size , 0 , SecurityConst . CSSM_KEYUSE_VERIFY , SecurityConst . CSSM_KEYATTR_EXTRACTABLE | SecurityConst . CSSM_KEYATTR_PERMANENT , SecurityConst . CSSM_KEYUSE_SIGN , SecurityConst . CSSM_KEYATTR_EXTRACTABLE | SecurityConst . CSSM_KEYATTR_PERMANENT , sec_access_ref , public_key_pointer , private_key_pointer )
        handle_sec_error ( result )
    else :
        cf_dict = CFHelpers . cf_dictionary_from_pairs ( [ ( Security . kSecAttrKeyType , key_type ) , ( Security . kSecAttrKeySizeInBits , CFHelpers . cf_number_from_integer ( key_size ) ) , ( Security . kSecAttrLabel , cf_string ) ] )
        result = Security . SecKeyGeneratePair ( cf_dict , public_key_pointer , private_key_pointer )
        handle_sec_error ( result )
    public_key_ref = unwrap ( public_key_pointer )
    private_key_ref = unwrap ( private_key_pointer )
    cf_data_public_pointer = new ( CoreFoundation , 'CFDataRef *' )
    result = Security . SecItemExport ( public_key_ref , 0 , 0 , null ( ) , cf_data_public_pointer )
    handle_sec_error ( result )
    cf_data_public = unwrap ( cf_data_public_pointer )
    public_key_bytes = CFHelpers . cf_data_to_bytes ( cf_data_public )
    cf_data_private_pointer = new ( CoreFoundation , 'CFDataRef *' )
    result = Security . SecItemExport ( private_key_ref , 0 , 0 , null ( ) , cf_data_private_pointer )
    handle_sec_error ( result )
    cf_data_private = unwrap ( cf_data_private_pointer )
    private_key_bytes = CFHelpers . cf_data_to_bytes ( cf_data_private )
    # Clean the new keys out of the keychain
    result = Security . SecKeychainItemDelete ( public_key_ref )
    handle_sec_error ( result )
    result = Security . SecKeychainItemDelete ( private_key_ref )
    handle_sec_error ( result )
finally :
    if cf_dict :
        CoreFoundation . CFRelease ( cf_dict )
    if public_key_ref :
        CoreFoundation . CFRelease ( public_key_ref )
    if private_key_ref :
        CoreFoundation . CFRelease ( private_key_ref )
    if cf_data_public :
        CoreFoundation . CFRelease ( cf_data_public )
    if cf_data_private :
        CoreFoundation . CFRelease ( cf_data_private )
    if cf_string :
        CoreFoundation . CFRelease ( cf_string )
    if sec_access_ref :
        CoreFoundation . CFRelease ( sec_access_ref )
return ( load_public_key ( public_key_bytes ) , load_private_key ( private_key_bytes ) )


signature:
def generate_dh_parameters ( bit_size ) :
    """Generates DH parameters for use with Diffie - Hellman key exchange . Returns
 a structure in the format of DHParameter defined in PKCS # 3 , which is also
 used by the OpenSSL dhparam tool .
 THIS CAN BE VERY TIME CONSUMING !
 : param bit _ size :
 The integer bit size of the parameters to generate . Must be between 512
 and 4096 , and divisible by 64 . Recommended secure value as of early 2016
 is 2048 , with an absolute minimum of 1024.
 : 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 :
 An asn1crypto . algos . DHParameters object . Use
 oscrypto . asymmetric . dump _ dh _ parameters ( ) to save to disk for usage with
 web servers ."""

implementation:
if not isinstance ( bit_size , int_types ) :
    raise TypeError ( pretty_message ( '''
            bit_size must be an integer, not %s
            ''' , type_name ( bit_size ) ) )
if bit_size < 512 :
    raise ValueError ( 'bit_size must be greater than or equal to 512' )
if bit_size > 4096 :
    raise ValueError ( 'bit_size must be less than or equal to 4096' )
if bit_size % 64 != 0 :
    raise ValueError ( 'bit_size must be a multiple of 64' )
public_key_ref = None
private_key_ref = None
cf_data_public = None
cf_data_private = None
cf_string = None
sec_access_ref = None
try :
    public_key_pointer = new ( Security , 'SecKeyRef *' )
    private_key_pointer = new ( Security , 'SecKeyRef *' )
    cf_string = CFHelpers . cf_string_from_unicode ( "Temporary key from oscrypto python library - safe to delete" )
    sec_access_ref_pointer = new ( Security , 'SecAccessRef *' )
    result = Security . SecAccessCreate ( cf_string , null ( ) , sec_access_ref_pointer )
    sec_access_ref = unwrap ( sec_access_ref_pointer )
    result = Security . SecKeyCreatePair ( null ( ) , SecurityConst . CSSM_ALGID_DH , bit_size , 0 , 0 , SecurityConst . CSSM_KEYATTR_EXTRACTABLE | SecurityConst . CSSM_KEYATTR_PERMANENT , 0 , SecurityConst . CSSM_KEYATTR_EXTRACTABLE | SecurityConst . CSSM_KEYATTR_PERMANENT , sec_access_ref , public_key_pointer , private_key_pointer )
    handle_sec_error ( result )
    public_key_ref = unwrap ( public_key_pointer )
    private_key_ref = unwrap ( private_key_pointer )
    cf_data_private_pointer = new ( CoreFoundation , 'CFDataRef *' )
    result = Security . SecItemExport ( private_key_ref , 0 , 0 , null ( ) , cf_data_private_pointer )
    handle_sec_error ( result )
    cf_data_private = unwrap ( cf_data_private_pointer )
    private_key_bytes = CFHelpers . cf_data_to_bytes ( cf_data_private )
    # Clean the new keys out of the keychain
    result = Security . SecKeychainItemDelete ( public_key_ref )
    handle_sec_error ( result )
    result = Security . SecKeychainItemDelete ( private_key_ref )
    handle_sec_error ( result )
    return algos . KeyExchangeAlgorithm . load ( private_key_bytes ) [ 'parameters' ]
finally :
    if public_key_ref :
        CoreFoundation . CFRelease ( public_key_ref )
    if private_key_ref :
        CoreFoundation . CFRelease ( private_key_ref )
    if cf_data_public :
        CoreFoundation . CFRelease ( cf_data_public )
    if cf_data_private :
        CoreFoundation . CFRelease ( cf_data_private )
    if cf_string :
        CoreFoundation . CFRelease ( cf_string )
    if sec_access_ref :
        CoreFoundation . CFRelease ( sec_access_ref )


signature:
def _load_x509 ( certificate ) :
    """Loads an ASN . 1 object of an x509 certificate into a Certificate object
 : param certificate :
 An asn1crypto . x509 . Certificate object
 : return :
 A Certificate object"""

implementation:
source = certificate . dump ( )
cf_source = None
try :
    cf_source = CFHelpers . cf_data_from_bytes ( source )
    sec_key_ref = Security . SecCertificateCreateWithData ( CoreFoundation . kCFAllocatorDefault , cf_source )
    return Certificate ( sec_key_ref , certificate )
finally :
    if cf_source :
        CoreFoundation . CFRelease ( cf_source )


signature:
def _load_key ( key_object ) :
    """Common code to load public and private keys into PublicKey and PrivateKey
 objects
 : param key _ object :
 An asn1crypto . keys . PublicKeyInfo or asn1crypto . keys . PrivateKeyInfo
 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 key is incompatible with the OS crypto library
 OSError - when an error is returned by the OS crypto library
 : return :
 A PublicKey or PrivateKey object"""

implementation:
if key_object . algorithm == 'ec' :
    curve_type , details = key_object . curve
    if curve_type != 'named' :
        raise AsymmetricKeyError ( 'OS X only supports EC keys using named curves' )
    if details not in set ( [ 'secp256r1' , 'secp384r1' , 'secp521r1' ] ) :
        raise AsymmetricKeyError ( pretty_message ( '''
                OS X only supports EC keys using the named curves secp256r1,
                secp384r1 and secp521r1
                ''' ) )
elif key_object . algorithm == 'dsa' and key_object . hash_algo == 'sha2' :
    raise AsymmetricKeyError ( pretty_message ( '''
            OS X only supports DSA keys based on SHA1 (2048 bits or less) - this
            key is based on SHA2 and is %s bits
            ''' , key_object . bit_size ) )
elif key_object . algorithm == 'dsa' and key_object . hash_algo is None :
    raise IncompleteAsymmetricKeyError ( pretty_message ( '''
            The DSA key does not contain the necessary p, q and g parameters
            and can not be used
            ''' ) )
if isinstance ( key_object , keys . PublicKeyInfo ) :
    source = key_object . dump ( )
    key_class = Security . kSecAttrKeyClassPublic
else :
    source = key_object . unwrap ( ) . dump ( )
    key_class = Security . kSecAttrKeyClassPrivate
cf_source = None
cf_dict = None
cf_output = None
try :
    cf_source = CFHelpers . cf_data_from_bytes ( source )
    key_type = { 'dsa' : Security . kSecAttrKeyTypeDSA , 'ec' : Security . kSecAttrKeyTypeECDSA , 'rsa' : Security . kSecAttrKeyTypeRSA , } [ key_object . algorithm ]
    cf_dict = CFHelpers . cf_dictionary_from_pairs ( [ ( Security . kSecAttrKeyType , key_type ) , ( Security . kSecAttrKeyClass , key_class ) , ( Security . kSecAttrCanSign , CoreFoundation . kCFBooleanTrue ) , ( Security . kSecAttrCanVerify , CoreFoundation . kCFBooleanTrue ) , ] )
    error_pointer = new ( CoreFoundation , 'CFErrorRef *' )
    sec_key_ref = Security . SecKeyCreateFromData ( cf_dict , cf_source , error_pointer )
    handle_cf_error ( error_pointer )
    if key_class == Security . kSecAttrKeyClassPublic :
        return PublicKey ( sec_key_ref , key_object )
    if key_class == Security . kSecAttrKeyClassPrivate :
        return PrivateKey ( sec_key_ref , key_object )
finally :
    if cf_source :
        CoreFoundation . CFRelease ( cf_source )
    if cf_dict :
        CoreFoundation . CFRelease ( cf_dict )
    if cf_output :
        CoreFoundation . CFRelease ( cf_output )


signature:
def rsa_pkcs1v15_encrypt ( certificate_or_public_key , data ) :
    """Encrypts a byte string using an RSA public key or certificate . Uses PKCS # 1
 v1.5 padding .
 : param certificate _ or _ public _ key :
 A PublicKey or Certificate object
 : param data :
 A byte string , with a maximum length 11 bytes less than the key length
 ( in bytes )
 : 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 encrypted data"""

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 ) ) )
key_length = certificate_or_public_key . byte_size
buffer = buffer_from_bytes ( key_length )
output_length = new ( Security , 'size_t *' , key_length )
result = Security . SecKeyEncrypt ( certificate_or_public_key . sec_key_ref , SecurityConst . kSecPaddingPKCS1 , data , len ( data ) , buffer , output_length )
handle_sec_error ( result )
return bytes_from_buffer ( buffer , deref ( output_length ) )


signature:
def rsa_pkcs1v15_decrypt ( private_key , ciphertext ) :
    """Decrypts a byte string using an RSA private key . Uses PKCS # 1 v1.5 padding .
 : param private _ key :
 A PrivateKey object
 : param ciphertext :
 A byte string of the encrypted data
 : 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 original plaintext"""

implementation:
if not isinstance ( private_key , PrivateKey ) :
    raise TypeError ( pretty_message ( '''
            private_key must an instance of the PrivateKey class, not %s
            ''' , type_name ( private_key ) ) )
if not isinstance ( ciphertext , byte_cls ) :
    raise TypeError ( pretty_message ( '''
            data must be a byte string, not %s
            ''' , type_name ( ciphertext ) ) )
key_length = private_key . byte_size
buffer = buffer_from_bytes ( key_length )
output_length = new ( Security , 'size_t *' , key_length )
if osx_version_info < ( 10 , 8 ) :
    padding = SecurityConst . kSecPaddingNone
else :
    padding = SecurityConst . kSecPaddingPKCS1
result = Security . SecKeyDecrypt ( private_key . sec_key_ref , padding , ciphertext , len ( ciphertext ) , buffer , output_length )
handle_sec_error ( result )
output = bytes_from_buffer ( buffer , deref ( output_length ) )
if osx_version_info < ( 10 , 8 ) :
    output = remove_pkcs1v15_encryption_padding ( key_length , output )
return output


signature:
def _encrypt ( certificate_or_public_key , data , padding ) :
    """Encrypts plaintext using an RSA public key or certificate
 : param certificate _ or _ public _ key :
 A Certificate or PublicKey object
 : param data :
 The plaintext - a byte string
 : param padding :
 The padding mode to use , specified as a kSecPadding * Key value
 : 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 padding :
    raise ValueError ( 'padding must be specified' )
cf_data = None
sec_transform = None
try :
    cf_data = CFHelpers . cf_data_from_bytes ( data )
    error_pointer = new ( CoreFoundation , 'CFErrorRef *' )
    sec_transform = Security . SecEncryptTransformCreate ( certificate_or_public_key . sec_key_ref , error_pointer )
    handle_cf_error ( error_pointer )
    if padding :
        Security . SecTransformSetAttribute ( sec_transform , Security . kSecPaddingKey , padding , error_pointer )
        handle_cf_error ( error_pointer )
    Security . SecTransformSetAttribute ( sec_transform , Security . kSecTransformInputAttributeName , cf_data , error_pointer )
    handle_cf_error ( error_pointer )
    ciphertext = Security . SecTransformExecute ( sec_transform , error_pointer )
    handle_cf_error ( error_pointer )
    return CFHelpers . cf_data_to_bytes ( ciphertext )
finally :
    if cf_data :
        CoreFoundation . CFRelease ( cf_data )
    if sec_transform :
        CoreFoundation . CFRelease ( sec_transform )


signature:
def _decrypt ( private_key , ciphertext , padding ) :
    """Decrypts RSA ciphertext using a private key
 : param private _ key :
 A PrivateKey object
 : param ciphertext :
 The ciphertext - a byte string
 : param padding :
 The padding mode to use , specified as a kSecPadding * Key value
 : 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 padding :
    raise ValueError ( 'padding must be specified' )
cf_data = None
sec_transform = None
try :
    cf_data = CFHelpers . cf_data_from_bytes ( ciphertext )
    error_pointer = new ( CoreFoundation , 'CFErrorRef *' )
    sec_transform = Security . SecDecryptTransformCreate ( private_key . sec_key_ref , error_pointer )
    handle_cf_error ( error_pointer )
    Security . SecTransformSetAttribute ( sec_transform , Security . kSecPaddingKey , padding , error_pointer )
    handle_cf_error ( error_pointer )
    Security . SecTransformSetAttribute ( sec_transform , Security . kSecTransformInputAttributeName , cf_data , error_pointer )
    handle_cf_error ( error_pointer )
    plaintext = Security . SecTransformExecute ( sec_transform , error_pointer )
    handle_cf_error ( error_pointer )
    return CFHelpers . cf_data_to_bytes ( plaintext )
finally :
    if cf_data :
        CoreFoundation . CFRelease ( cf_data )
    if sec_transform :
        CoreFoundation . CFRelease ( sec_transform )


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 " , " sha224 " , " 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 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 certificate_or_public_key . algorithm != 'rsa' :
    raise ValueError ( 'The key specified is not an RSA public key' )
hash_length = { 'sha1' : 20 , 'sha224' : 28 , 'sha256' : 32 , 'sha384' : 48 , 'sha512' : 64 } . get ( hash_algorithm , 0 )
key_length = certificate_or_public_key . byte_size
buffer = buffer_from_bytes ( key_length )
output_length = new ( Security , 'size_t *' , key_length )
result = Security . SecKeyEncrypt ( certificate_or_public_key . sec_key_ref , SecurityConst . kSecPaddingNone , signature , len ( signature ) , buffer , output_length )
handle_sec_error ( result )
plaintext = bytes_from_buffer ( buffer , deref ( output_length ) )
if not verify_pss_padding ( hash_algorithm , hash_length , certificate_or_public_key . bit_size , data , plaintext ) :
    raise SignatureError ( 'Signature is invalid' )