def _handshake(self):
session_context = None<EOL>ssl_policy_ref = None<EOL>crl_search_ref = None<EOL>crl_policy_ref = None<EOL>ocsp_search_ref = None<EOL>ocsp_policy_ref = None<EOL>policy_array_ref = None<EOL>try:<EOL><INDENT>if osx_version_info < (<NUM_LIT:10>, <NUM_LIT:8>):<EOL><INDENT>session_context_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLNewContext(False, session_context_pointer)<EOL>handle_sec_error(result)<EOL>session_context = unwrap(session_context_pointer)<EOL><DEDENT>else:<EOL><INDENT>session_context = Security.SSLCreateContext(<EOL>null(),<EOL>SecurityConst.kSSLClientSide,<EOL>SecurityConst.kSSLStreamType<EOL>)<EOL><DEDENT>result = Security.SSLSetIOFuncs(<EOL>session_context,<EOL>_read_callback_pointer,<EOL>_write_callback_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>self._connection_id = id(self) % <NUM_LIT><EOL>_connection_refs[self._connection_id] = self<EOL>_socket_refs[self._connection_id] = self._socket<EOL>result = Security.SSLSetConnection(session_context, self._connection_id)<EOL>handle_sec_error(result)<EOL>utf8_domain = self._hostname.encode('<STR_LIT:utf-8>')<EOL>result = Security.SSLSetPeerDomainName(<EOL>session_context,<EOL>utf8_domain,<EOL>len(utf8_domain)<EOL>)<EOL>handle_sec_error(result)<EOL>if osx_version_info >= (<NUM_LIT:10>, <NUM_LIT:10>):<EOL><INDENT>disable_auto_validation = self._session._manual_validation or self._session._extra_trust_roots<EOL>explicit_validation = (not self._session._manual_validation) and self._session._extra_trust_roots<EOL><DEDENT>else:<EOL><INDENT>disable_auto_validation = True<EOL>explicit_validation = not self._session._manual_validation<EOL><DEDENT>if osx_version_info < (<NUM_LIT:10>, <NUM_LIT:8>):<EOL><INDENT>for protocol in ['<STR_LIT>', '<STR_LIT>', '<STR_LIT>']:<EOL><INDENT>protocol_const = _PROTOCOL_STRING_CONST_MAP[protocol]<EOL>enabled = protocol in self._session._protocols<EOL>result = Security.SSLSetProtocolVersionEnabled(<EOL>session_context,<EOL>protocol_const,<EOL>enabled<EOL>)<EOL>handle_sec_error(result)<EOL><DEDENT>if disable_auto_validation:<EOL><INDENT>result = Security.SSLSetEnableCertVerify(session_context, False)<EOL>handle_sec_error(result)<EOL><DEDENT><DEDENT>else:<EOL><INDENT>protocol_consts = [_PROTOCOL_STRING_CONST_MAP[protocol] for protocol in self._session._protocols]<EOL>min_protocol = min(protocol_consts)<EOL>max_protocol = max(protocol_consts)<EOL>result = Security.SSLSetProtocolVersionMin(<EOL>session_context,<EOL>min_protocol<EOL>)<EOL>handle_sec_error(result)<EOL>result = Security.SSLSetProtocolVersionMax(<EOL>session_context,<EOL>max_protocol<EOL>)<EOL>handle_sec_error(result)<EOL>if disable_auto_validation:<EOL><INDENT>result = Security.SSLSetSessionOption(<EOL>session_context,<EOL>SecurityConst.kSSLSessionOptionBreakOnServerAuth,<EOL>True<EOL>)<EOL>handle_sec_error(result)<EOL><DEDENT><DEDENT>supported_ciphers_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLGetNumberSupportedCiphers(session_context, supported_ciphers_pointer)<EOL>handle_sec_error(result)<EOL>supported_ciphers = deref(supported_ciphers_pointer)<EOL>cipher_buffer = buffer_from_bytes(supported_ciphers * <NUM_LIT:4>)<EOL>supported_cipher_suites_pointer = cast(Security, '<STR_LIT>', cipher_buffer)<EOL>result = Security.SSLGetSupportedCiphers(<EOL>session_context,<EOL>supported_cipher_suites_pointer,<EOL>supported_ciphers_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>supported_ciphers = deref(supported_ciphers_pointer)<EOL>supported_cipher_suites = array_from_pointer(<EOL>Security,<EOL>'<STR_LIT>',<EOL>supported_cipher_suites_pointer,<EOL>supported_ciphers<EOL>)<EOL>good_ciphers = []<EOL>for supported_cipher_suite in supported_cipher_suites:<EOL><INDENT>cipher_suite = int_to_bytes(supported_cipher_suite, width=<NUM_LIT:2>)<EOL>cipher_suite_name = CIPHER_SUITE_MAP.get(cipher_suite, cipher_suite)<EOL>good_cipher = _cipher_blacklist_regex.search(cipher_suite_name) is None<EOL>if good_cipher:<EOL><INDENT>good_ciphers.append(supported_cipher_suite)<EOL><DEDENT><DEDENT>num_good_ciphers = len(good_ciphers)<EOL>good_ciphers_array = new(Security, '<STR_LIT>', num_good_ciphers)<EOL>array_set(good_ciphers_array, good_ciphers)<EOL>good_ciphers_pointer = cast(Security, '<STR_LIT>', good_ciphers_array)<EOL>result = Security.SSLSetEnabledCiphers(<EOL>session_context,<EOL>good_ciphers_pointer,<EOL>num_good_ciphers<EOL>)<EOL>handle_sec_error(result)<EOL>peer_id = self._session._peer_id + self._hostname.encode('<STR_LIT:utf-8>')<EOL>result = Security.SSLSetPeerID(session_context, peer_id, len(peer_id))<EOL>handle_sec_error(result)<EOL>handshake_result = Security.SSLHandshake(session_context)<EOL>if self._exception is not None:<EOL><INDENT>exception = self._exception<EOL>self._exception = None<EOL>raise exception<EOL><DEDENT>while handshake_result == SecurityConst.errSSLWouldBlock:<EOL><INDENT>handshake_result = Security.SSLHandshake(session_context)<EOL>if self._exception is not None:<EOL><INDENT>exception = self._exception<EOL>self._exception = None<EOL>raise exception<EOL><DEDENT><DEDENT>if osx_version_info < (<NUM_LIT:10>, <NUM_LIT:8>) and osx_version_info >= (<NUM_LIT:10>, <NUM_LIT:7>):<EOL><INDENT>do_validation = explicit_validation and handshake_result == <NUM_LIT:0><EOL><DEDENT>else:<EOL><INDENT>do_validation = explicit_validation and handshake_result == SecurityConst.errSSLServerAuthCompleted<EOL><DEDENT>if do_validation:<EOL><INDENT>trust_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLCopyPeerTrust(<EOL>session_context,<EOL>trust_ref_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>trust_ref = unwrap(trust_ref_pointer)<EOL>cf_string_hostname = CFHelpers.cf_string_from_unicode(self._hostname)<EOL>ssl_policy_ref = Security.SecPolicyCreateSSL(True, cf_string_hostname)<EOL>result = CoreFoundation.CFRelease(cf_string_hostname)<EOL>handle_cf_error(result)<EOL>ocsp_oid_pointer = struct(Security, '<STR_LIT>')<EOL>ocsp_oid = unwrap(ocsp_oid_pointer)<EOL>ocsp_oid.Length = len(SecurityConst.APPLE_TP_REVOCATION_OCSP)<EOL>ocsp_oid_buffer = buffer_from_bytes(SecurityConst.APPLE_TP_REVOCATION_OCSP)<EOL>ocsp_oid.Data = cast(Security, '<STR_LIT>', ocsp_oid_buffer)<EOL>ocsp_search_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecPolicySearchCreate(<EOL>SecurityConst.CSSM_CERT_X_509v3,<EOL>ocsp_oid_pointer,<EOL>null(),<EOL>ocsp_search_ref_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>ocsp_search_ref = unwrap(ocsp_search_ref_pointer)<EOL>ocsp_policy_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecPolicySearchCopyNext(ocsp_search_ref, ocsp_policy_ref_pointer)<EOL>handle_sec_error(result)<EOL>ocsp_policy_ref = unwrap(ocsp_policy_ref_pointer)<EOL>ocsp_struct_pointer = struct(Security, '<STR_LIT>')<EOL>ocsp_struct = unwrap(ocsp_struct_pointer)<EOL>ocsp_struct.Version = SecurityConst.CSSM_APPLE_TP_OCSP_OPTS_VERSION<EOL>ocsp_struct.Flags = (<EOL>SecurityConst.CSSM_TP_ACTION_OCSP_DISABLE_NET |<EOL>SecurityConst.CSSM_TP_ACTION_OCSP_CACHE_READ_DISABLE<EOL>)<EOL>ocsp_struct_bytes = struct_bytes(ocsp_struct_pointer)<EOL>cssm_data_pointer = struct(Security, '<STR_LIT>')<EOL>cssm_data = unwrap(cssm_data_pointer)<EOL>cssm_data.Length = len(ocsp_struct_bytes)<EOL>ocsp_struct_buffer = buffer_from_bytes(ocsp_struct_bytes)<EOL>cssm_data.Data = cast(Security, '<STR_LIT>', ocsp_struct_buffer)<EOL>result = Security.SecPolicySetValue(ocsp_policy_ref, cssm_data_pointer)<EOL>handle_sec_error(result)<EOL>crl_oid_pointer = struct(Security, '<STR_LIT>')<EOL>crl_oid = unwrap(crl_oid_pointer)<EOL>crl_oid.Length = len(SecurityConst.APPLE_TP_REVOCATION_CRL)<EOL>crl_oid_buffer = buffer_from_bytes(SecurityConst.APPLE_TP_REVOCATION_CRL)<EOL>crl_oid.Data = cast(Security, '<STR_LIT>', crl_oid_buffer)<EOL>crl_search_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecPolicySearchCreate(<EOL>SecurityConst.CSSM_CERT_X_509v3,<EOL>crl_oid_pointer,<EOL>null(),<EOL>crl_search_ref_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>crl_search_ref = unwrap(crl_search_ref_pointer)<EOL>crl_policy_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecPolicySearchCopyNext(crl_search_ref, crl_policy_ref_pointer)<EOL>handle_sec_error(result)<EOL>crl_policy_ref = unwrap(crl_policy_ref_pointer)<EOL>crl_struct_pointer = struct(Security, '<STR_LIT>')<EOL>crl_struct = unwrap(crl_struct_pointer)<EOL>crl_struct.Version = SecurityConst.CSSM_APPLE_TP_CRL_OPTS_VERSION<EOL>crl_struct.CrlFlags = <NUM_LIT:0><EOL>crl_struct_bytes = struct_bytes(crl_struct_pointer)<EOL>cssm_data_pointer = struct(Security, '<STR_LIT>')<EOL>cssm_data = unwrap(cssm_data_pointer)<EOL>cssm_data.Length = len(crl_struct_bytes)<EOL>crl_struct_buffer = buffer_from_bytes(crl_struct_bytes)<EOL>cssm_data.Data = cast(Security, '<STR_LIT>', crl_struct_buffer)<EOL>result = Security.SecPolicySetValue(crl_policy_ref, cssm_data_pointer)<EOL>handle_sec_error(result)<EOL>policy_array_ref = CFHelpers.cf_array_from_list([<EOL>ssl_policy_ref,<EOL>crl_policy_ref,<EOL>ocsp_policy_ref<EOL>])<EOL>result = Security.SecTrustSetPolicies(trust_ref, policy_array_ref)<EOL>handle_sec_error(result)<EOL>if self._session._extra_trust_roots:<EOL><INDENT>ca_cert_refs = []<EOL>ca_certs = []<EOL>for cert in self._session._extra_trust_roots:<EOL><INDENT>ca_cert = load_certificate(cert)<EOL>ca_certs.append(ca_cert)<EOL>ca_cert_refs.append(ca_cert.sec_certificate_ref)<EOL><DEDENT>result = Security.SecTrustSetAnchorCertificatesOnly(trust_ref, False)<EOL>handle_sec_error(result)<EOL>array_ref = CFHelpers.cf_array_from_list(ca_cert_refs)<EOL>result = Security.SecTrustSetAnchorCertificates(trust_ref, array_ref)<EOL>handle_sec_error(result)<EOL><DEDENT>result_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecTrustEvaluate(trust_ref, result_pointer)<EOL>handle_sec_error(result)<EOL>trust_result_code = deref(result_pointer)<EOL>invalid_chain_error_codes = set([<EOL>SecurityConst.kSecTrustResultProceed,<EOL>SecurityConst.kSecTrustResultUnspecified<EOL>])<EOL>if trust_result_code not in invalid_chain_error_codes:<EOL><INDENT>handshake_result = SecurityConst.errSSLXCertChainInvalid<EOL><DEDENT>else:<EOL><INDENT>handshake_result = Security.SSLHandshake(session_context)<EOL>while handshake_result == SecurityConst.errSSLWouldBlock:<EOL><INDENT>handshake_result = Security.SSLHandshake(session_context)<EOL><DEDENT><DEDENT><DEDENT>self._done_handshake = True<EOL>handshake_error_codes = set([<EOL>SecurityConst.errSSLXCertChainInvalid,<EOL>SecurityConst.errSSLCertExpired,<EOL>SecurityConst.errSSLCertNotYetValid,<EOL>SecurityConst.errSSLUnknownRootCert,<EOL>SecurityConst.errSSLNoRootCert,<EOL>SecurityConst.errSSLHostNameMismatch,<EOL>SecurityConst.errSSLInternal,<EOL>])<EOL>if handshake_result in handshake_error_codes:<EOL><INDENT>trust_ref_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLCopyPeerTrust(<EOL>session_context,<EOL>trust_ref_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>trust_ref = unwrap(trust_ref_pointer)<EOL>result_code_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SecTrustGetCssmResultCode(trust_ref, result_code_pointer)<EOL>result_code = deref(result_code_pointer)<EOL>chain = extract_chain(self._server_hello)<EOL>self_signed = False<EOL>revoked = False<EOL>expired = False<EOL>not_yet_valid = False<EOL>no_issuer = False<EOL>cert = None<EOL>bad_hostname = False<EOL>if chain:<EOL><INDENT>cert = chain[<NUM_LIT:0>]<EOL>oscrypto_cert = load_certificate(cert)<EOL>self_signed = oscrypto_cert.self_signed<EOL>revoked = result_code == SecurityConst.CSSMERR_TP_CERT_REVOKED<EOL>no_issuer = not self_signed and result_code == SecurityConst.CSSMERR_TP_NOT_TRUSTED<EOL>expired = result_code == SecurityConst.CSSMERR_TP_CERT_EXPIRED<EOL>not_yet_valid = result_code == SecurityConst.CSSMERR_TP_CERT_NOT_VALID_YET<EOL>bad_hostname = result_code == SecurityConst.CSSMERR_APPLETP_HOSTNAME_MISMATCH<EOL>if osx_version_info >= (<NUM_LIT:10>, <NUM_LIT:12>):<EOL><INDENT>validity = cert['<STR_LIT>']['<STR_LIT>']<EOL>not_before = validity['<STR_LIT>'].chosen.native<EOL>not_after = validity['<STR_LIT>'].chosen.native<EOL>utcnow = datetime.datetime.now(timezone.utc)<EOL>expired = not_after < utcnow<EOL>not_yet_valid = not_before > utcnow<EOL><DEDENT><DEDENT>if chain and chain[<NUM_LIT:0>].hash_algo in set(['<STR_LIT>', '<STR_LIT>']):<EOL><INDENT>raise_weak_signature(chain[<NUM_LIT:0>])<EOL><DEDENT>if revoked:<EOL><INDENT>raise_revoked(cert)<EOL><DEDENT>if bad_hostname:<EOL><INDENT>raise_hostname(cert, self._hostname)<EOL><DEDENT>elif expired or not_yet_valid:<EOL><INDENT>raise_expired_not_yet_valid(cert)<EOL><DEDENT>elif no_issuer:<EOL><INDENT>raise_no_issuer(cert)<EOL><DEDENT>elif self_signed:<EOL><INDENT>raise_self_signed(cert)<EOL><DEDENT>if detect_client_auth_request(self._server_hello):<EOL><INDENT>raise_client_auth()<EOL><DEDENT>raise_verification(cert)<EOL><DEDENT>if handshake_result == SecurityConst.errSSLPeerHandshakeFail:<EOL><INDENT>if detect_client_auth_request(self._server_hello):<EOL><INDENT>raise_client_auth()<EOL><DEDENT>raise_handshake()<EOL><DEDENT>if handshake_result == SecurityConst.errSSLWeakPeerEphemeralDHKey:<EOL><INDENT>raise_dh_params()<EOL><DEDENT>if handshake_result == SecurityConst.errSSLPeerProtocolVersion:<EOL><INDENT>raise_protocol_version()<EOL><DEDENT>if handshake_result in set([SecurityConst.errSSLRecordOverflow, SecurityConst.errSSLProtocol]):<EOL><INDENT>self._server_hello += _read_remaining(self._socket)<EOL>raise_protocol_error(self._server_hello)<EOL><DEDENT>if handshake_result in set([SecurityConst.errSSLClosedNoNotify, SecurityConst.errSSLClosedAbort]):<EOL><INDENT>if not self._done_handshake:<EOL><INDENT>self._server_hello += _read_remaining(self._socket)<EOL><DEDENT>if detect_other_protocol(self._server_hello):<EOL><INDENT>raise_protocol_error(self._server_hello)<EOL><DEDENT>raise_disconnection()<EOL><DEDENT>if osx_version_info < (<NUM_LIT:10>, <NUM_LIT:10>):<EOL><INDENT>dh_params_length = get_dh_params_length(self._server_hello)<EOL>if dh_params_length is not None and dh_params_length < <NUM_LIT>:<EOL><INDENT>raise_dh_params()<EOL><DEDENT><DEDENT>would_block = handshake_result == SecurityConst.errSSLWouldBlock<EOL>server_auth_complete = handshake_result == SecurityConst.errSSLServerAuthCompleted<EOL>manual_validation = self._session._manual_validation and server_auth_complete<EOL>if not would_block and not manual_validation:<EOL><INDENT>handle_sec_error(handshake_result, TLSError)<EOL><DEDENT>self._session_context = session_context<EOL>protocol_const_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLGetNegotiatedProtocolVersion(<EOL>session_context,<EOL>protocol_const_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>protocol_const = deref(protocol_const_pointer)<EOL>self._protocol = _PROTOCOL_CONST_STRING_MAP[protocol_const]<EOL>cipher_int_pointer = new(Security, '<STR_LIT>')<EOL>result = Security.SSLGetNegotiatedCipher(<EOL>session_context,<EOL>cipher_int_pointer<EOL>)<EOL>handle_sec_error(result)<EOL>cipher_int = deref(cipher_int_pointer)<EOL>cipher_bytes = int_to_bytes(cipher_int, width=<NUM_LIT:2>)<EOL>self._cipher_suite = CIPHER_SUITE_MAP.get(cipher_bytes, cipher_bytes)<EOL>session_info = parse_session_info(<EOL>self._server_hello,<EOL>self._client_hello<EOL>)<EOL>self._compression = session_info['<STR_LIT>']<EOL>self._session_id = session_info['<STR_LIT>']<EOL>self._session_ticket = session_info['<STR_LIT>']<EOL><DEDENT>except (OSError, socket_.error):<EOL><INDENT>if session_context:<EOL><INDENT>if osx_version_info < (<NUM_LIT:10>, <NUM_LIT:8>):<EOL><INDENT>result = Security.SSLDisposeContext(session_context)<EOL>handle_sec_error(result)<EOL><DEDENT>else:<EOL><INDENT>result = CoreFoundation.CFRelease(session_context)<EOL>handle_cf_error(result)<EOL><DEDENT><DEDENT>self._session_context = None<EOL>self.close()<EOL>raise<EOL><DEDENT>finally:<EOL><INDENT>if ssl_policy_ref:<EOL><INDENT>result = CoreFoundation.CFRelease(ssl_policy_ref)<EOL>handle_cf_error(result)<EOL>ssl_policy_ref = None<EOL><DEDENT>if crl_policy_ref:<EOL><INDENT>result = CoreFoundation.CFRelease(crl_policy_ref)<EOL>handle_cf_error(result)<EOL>crl_policy_ref = None<EOL><DEDENT>if ocsp_policy_ref:<EOL><INDENT>result = CoreFoundation.CFRelease(ocsp_policy_ref)<EOL>handle_cf_error(result)<EOL>ocsp_policy_ref = None<EOL><DEDENT>if policy_array_ref:<EOL><INDENT>result = CoreFoundation.CFRelease(policy_array_ref)<EOL>handle_cf_error(result)<EOL>policy_array_ref = None<EOL><DEDENT><DEDENT>
Perform an initial TLS handshake