我们从Python开源项目中，提取了以下11个代码示例，用于说明如何使用httplib2.SSLHandshakeError()。
def getAPIversionHttpService(api): hasLocalJSON = API.hasLocalJSON(api) api, version, api_version, cred_family = API.getVersion(api) httpObj = httplib2.Http(disable_ssl_certificate_validation=GC.Values[GC.NO_VERIFY_SSL], cache=GM.Globals[GM.CACHE_DIR]) if not hasLocalJSON: retries = 3 for n in range(1, retries+1): try: service = googleapiclient.discovery.build(api, version, http=httpObj, cache_discovery=False) if GM.Globals[GM.CACHE_DISCOVERY_ONLY]: httpObj.cache = None return (api_version, httpObj, service, cred_family) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e)) except googleapiclient.errors.UnknownApiNameOrVersion as e: systemErrorExit(GOOGLE_API_ERROR_RC, Msg.UNKNOWN_API_OR_VERSION.format(str(e), __author__)) except (googleapiclient.errors.InvalidJsonError, KeyError, ValueError): httpObj.cache = None if n != retries: waitOnFailure(n, retries, INVALID_JSON_RC, Msg.INVALID_JSON_INFORMATION) continue systemErrorExit(INVALID_JSON_RC, Msg.INVALID_JSON_INFORMATION) except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue systemErrorExit(SOCKET_ERROR_RC, errMsg) disc_file, discovery = readDiscoveryFile(api_version) try: service = googleapiclient.discovery.build_from_document(discovery, http=httpObj) if GM.Globals[GM.CACHE_DISCOVERY_ONLY]: httpObj.cache = None return (api_version, httpObj, service, cred_family) except (KeyError, ValueError): invalidDiscoveryJsonExit(disc_file)
def testSslCertValidationWithSelfSignedCaCert(self): if sys.version_info >= (2, 7): other_ca_certs = os.path.join( os.path.dirname(os.path.abspath(httplib2.__file__ )), 'test', 'other_cacerts.txt') http = httplib2.Http(ca_certs=other_ca_certs) if sys.platform != 'darwin': with self.assertRaises(httplib2.SSLHandshakeError): http.request('https://www.google.com/', 'GET')
def testSslProtocolV3AndShouldFailDueToPoodle(self): http = httplib2.Http(ssl_version=ssl.PROTOCOL_SSLv3) urls = ['https://www.amazon.com', 'https://www.apple.com', 'https://www.twitter.com'] for url in urls: if sys.version_info >= (2, 7): with self.assertRaises(httplib2.SSLHandshakeError): http.request(url) try: http.request(url) except httplib2.SSLHandshakeError as e: self.assertTrue('sslv3 alert handshake failure' in str(e))
def _send_request(self, uri, method, headers=None, body=None): """Do send request to target URL and validate SSL cert by default. If validation failed, disable it and try again.""" if self._connection is None: self._connection = self._build_http_connection( proxy_info=self._proxy_info, disable_ssl_cert_validation=False) try: return self._connection.request( uri, body=body, method=method, headers=headers ) except SSLHandshakeError: _logger.warning( "[SSL: CERTIFICATE_VERIFY_FAILED] certificate verification failed. " "The certificate of the https server [%s] is not trusted, " "this add-on will proceed to connect with this certificate. " "You may need to check the certificate and " "refer to the documentation and add it to the trust list. %s", uri, traceback.format_exc() ) self._connection = self._build_http_connection( proxy_info=self._proxy_info, disable_ssl_cert_validation=True ) return self._connection.request( uri, body=body, method=method, headers=headers )
def callGData(service, function, soft_errors=False, throw_errors=None, retry_errors=None, **kwargs): import gdata.apps.service if throw_errors is None: throw_errors = [] if retry_errors is None: retry_errors = [] all_retry_errors = GDATA.NON_TERMINATING_ERRORS+retry_errors method = getattr(service, function) retries = 10 for n in range(1, retries+1): try: return method(**kwargs) except gdata.apps.service.AppsForYourDomainException as e: error_code, error_message = checkGDataError(e, service) if (n != retries) and (error_code in all_retry_errors): waitOnFailure(n, retries, error_code, error_message) continue if error_code in throw_errors: if error_code in GDATA.ERROR_CODE_EXCEPTION_MAP: raise GDATA.ERROR_CODE_EXCEPTION_MAP[error_code](error_message) raise if soft_errors: stderrErrorMsg(u'{0} - {1}{2}'.format(error_code, error_message, [u'', u': Giving up.'][n > 1])) return None if error_code == GDATA.INSUFFICIENT_PERMISSIONS: APIAccessDeniedExit() systemErrorExit(GOOGLE_API_ERROR_RC, u'{0} - {1}'.format(error_code, error_message)) except oauth2client.client.AccessTokenRefreshError as e: handleOAuthTokenError(str(e), GDATA.SERVICE_NOT_APPLICABLE in throw_errors) raise GDATA.ERROR_CODE_EXCEPTION_MAP[GDATA.SERVICE_NOT_APPLICABLE](str(e)) except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue if soft_errors: writeStderr(convertUTF8(u'\n{0}{1} - Giving up.\n'.format(ERROR_PREFIX, errMsg))) return None systemErrorExit(SOCKET_ERROR_RC, errMsg) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e))
def callGAPI(service, function, silent_errors=False, soft_errors=False, throw_reasons=None, retry_reasons=None, **kwargs): if throw_reasons is None: throw_reasons = [] if retry_reasons is None: retry_reasons = [] all_retry_reasons = GAPI.DEFAULT_RETRY_REASONS+retry_reasons method = getattr(service, function) retries = 10 svcparms = dict(kwargs.items()+GM.Globals[GM.EXTRA_ARGS_LIST]) for n in range(1, retries+1): try: return method(**svcparms).execute() except googleapiclient.errors.HttpError as e: http_status, reason, message = checkGAPIError(e, soft_errors=soft_errors, silent_errors=silent_errors, retryOnHttpError=n < 3, service=service) if http_status == -1: continue if http_status == 0: return None if (n != retries) and (reason in all_retry_reasons): waitOnFailure(n, retries, reason, message) continue if reason in throw_reasons: if reason in GAPI.REASON_EXCEPTION_MAP: raise GAPI.REASON_EXCEPTION_MAP[reason](message) raise e if soft_errors: stderrErrorMsg(u'{0}: {1} - {2}{3}'.format(http_status, reason, message, [u'', u': Giving up.'][n > 1])) return None if reason == GAPI.INSUFFICIENT_PERMISSIONS: APIAccessDeniedExit() systemErrorExit(HTTP_ERROR_RC, formatHTTPError(http_status, reason, message)) except oauth2client.client.AccessTokenRefreshError as e: handleOAuthTokenError(str(e), GAPI.SERVICE_NOT_AVAILABLE in throw_reasons) raise GAPI.REASON_EXCEPTION_MAP[GAPI.SERVICE_NOT_AVAILABLE](str(e)) except httplib2.CertificateValidationUnsupported: noPythonSSLExit() except (http_client.ResponseNotReady, httplib2.SSLHandshakeError, socket.error) as e: errMsg = u'Connection error: {0}'.format(str(e) or repr(e)) if n != retries: waitOnFailure(n, retries, SOCKET_ERROR_RC, errMsg) continue if soft_errors: writeStderr(convertUTF8(u'\n{0}{1} - Giving up.\n'.format(ERROR_PREFIX, errMsg))) return None systemErrorExit(SOCKET_ERROR_RC, errMsg) except ValueError as e: if service._http.cache is not None: service._http.cache = None continue systemErrorExit(GOOGLE_API_ERROR_RC, str(e)) except TypeError as e: systemErrorExit(GOOGLE_API_ERROR_RC, str(e)) except httplib2.ServerNotFoundError as e: systemErrorExit(NETWORK_ERROR_RC, str(e))
