我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用ssl.SSLError()。
def _tcp_proc(self, addrinfo, task=None): """ Internal use only. """ task.set_daemon() sock = addrinfo.tcp_sock while 1: try: conn, addr = yield sock.accept() except ssl.SSLError as err: logger.debug('SSL connection failed: %s', str(err)) continue except GeneratorExit: break except: logger.debug(traceback.format_exc()) continue SysTask(self._tcp_conn_proc, conn, addrinfo)
def _assert_no_error(error, exception_class=None): """ Checks the return code and throws an exception if there is an error to report """ if error == 0: return cf_error_string = Security.SecCopyErrorMessageString(error, None) output = _cf_string_to_unicode(cf_error_string) CoreFoundation.CFRelease(cf_error_string) if output is None or output == u'': output = u'OSStatus %s' % error if exception_class is None: exception_class = ssl.SSLError raise exception_class(output)
def handle_read(self): try: while True: buf = self.recv(self.in_buffer_size) self._iobuf.write(buf) if len(buf) < self.in_buffer_size: break except socket.error as err: if ssl and isinstance(err, ssl.SSLError): if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): self.defunct(err) return elif err.args[0] not in NONBLOCKING: self.defunct(err) return if self._iobuf.tell(): self.process_io_buffer() if not self._requests and not self.is_control_connection: self._readable = False
def connect(self): """Connect to Mongo and return a new (connected) socket. Note that the pool does not keep a reference to the socket -- you must call return_socket() when you're done with it. """ sock = self.create_connection() hostname = self.pair[0] if self.use_ssl: try: sock = ssl.wrap_socket(sock, certfile=self.ssl_certfile, keyfile=self.ssl_keyfile, ca_certs=self.ssl_ca_certs, cert_reqs=self.ssl_cert_reqs) if self.ssl_cert_reqs: match_hostname(sock.getpeercert(), hostname) except ssl.SSLError: sock.close() raise ConnectionFailure("SSL handshake failed. MongoDB may " "not be configured with SSL support.") sock.settimeout(self.net_timeout) return SocketInfo(sock, self.pool_id, hostname)
def get_ssl_certificate(self, binary_form=False): """Returns the client's SSL certificate, if any. To use client certificates, the HTTPServer's `ssl.SSLContext.verify_mode` field must be set, e.g.:: ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_ctx.load_cert_chain("foo.crt", "foo.key") ssl_ctx.load_verify_locations("cacerts.pem") ssl_ctx.verify_mode = ssl.CERT_REQUIRED server = HTTPServer(app, ssl_options=ssl_ctx) By default, the return value is a dictionary (or None, if no client certificate is present). If ``binary_form`` is true, a DER-encoded form of the certificate is returned instead. See SSLSocket.getpeercert() in the standard library for more details. http://docs.python.org/library/ssl.html#sslsocket-objects """ try: return self.connection.stream.socket.getpeercert( binary_form=binary_form) except SSLError: return None
def _check_ssl_cert(self, cert, key): # Check SSL-Certificate with openssl, if possible try: exit_code = subprocess.call( ["openssl", "x509", "-text", "-noout", "-in", cert], stdout=open(os.devnull, 'wb'), stderr=subprocess.STDOUT) except OSError: exit_code = 0 if exit_code is 0: try: self.httpd.socket = ssl.wrap_socket( self.httpd.socket, certfile=cert, keyfile=key, server_side=True) except ssl.SSLError as error: self.logger.exception('Failed to init SSL socket') raise TelegramError(str(error)) else: raise TelegramError('SSL Certificate invalid')
def close(self): """Closes the connection to the email server.""" if self.connection is None: return try: try: self.connection.quit() except (ssl.SSLError, smtplib.SMTPServerDisconnected): # This happens when calling quit() on a TLS connection # sometimes, or when the connection was already disconnected # by the server. self.connection.close() except smtplib.SMTPException: if self.fail_silently: return raise finally: self.connection = None
def recv_and_split_blocks(socket, length): buff = '' try: while True: c = socket.recv(1) if c == '': break buff += c if (len(buff) % length) == 0: yield buff buff = '' except ssl.SSLError as err: print("SSL error: {}".format(err.reason)) pass except Exception as err: #print(traceback.format_exc()) pass if len(buff) != 0: yield buff
def testSslCertValidation(self): # Test that we get an ssl.SSLError when specifying a non-existent CA # certs file. http = httplib2.Http(ca_certs='/nosuchfile') self.assertRaises(IOError, http.request, "https://www.google.com/", "GET") # Test that we get a SSLHandshakeError if we try to access # https://www.google.com, using a CA cert file that doesn't contain # the CA Google uses (i.e., simulating a cert that's not signed by a # trusted CA). 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) self.assertRaises(ssl.SSLError, http.request,"https://www.google.com/", "GET")
def data_received(self, data): """Called when some SSL data is received. The argument is a bytes object. """ try: ssldata, appdata = self._sslpipe.feed_ssldata(data) except ssl.SSLError as e: if self._loop.get_debug(): logger.warning('%r: SSL error %s (reason %s)', self, e.errno, e.reason) self._abort() return for chunk in ssldata: self._transport.write(chunk) for chunk in appdata: if chunk: self._app_protocol.data_received(chunk) else: self._start_shutdown() break
def websocket_safe_read(self): """Returns data if available, otherwise ''. Newlines indicate multiple messages """ data = '' while True: try: data += '{0}\n'.format(self.websocket.recv()) except WebSocketException as e: if isinstance(e, WebSocketConnectionClosedException): logger.warning('lost websocket connection, try to reconnect now') else: logger.warning('websocket exception: %s', e) self.reconnect() except Exception as e: if isinstance(e, SSLError) and e.errno == 2: pass else: logger.warning('Exception in websocket_safe_read: %s', e) return data.rstrip()
def ssl_wrap_socket(sock, keyfile=None, certfile=None, cert_reqs=CERT_NONE, ca_certs=None, server_hostname=None, ssl_version=PROTOCOL_SSLv23): """ All arguments except `server_hostname` have the same meaning as for :func:`ssl.wrap_socket` :param server_hostname: Hostname of the expected certificate """ context = SSLContext(ssl_version) context.verify_mode = cert_reqs if ca_certs: try: context.load_verify_locations(ca_certs) except TypeError as e: # Reraise as SSLError # FIXME: This block needs a test. raise SSLError(e) if certfile: # FIXME: This block needs a test. context.load_cert_chain(certfile, keyfile) if HAS_SNI: # Platform-specific: OpenSSL with enabled SNI return context.wrap_socket(sock, server_hostname=server_hostname) return context.wrap_socket(sock)
def wrap_socket(self, sock): try: if self.clientcert_req: ca_certs = self.interface[4] cert_reqs = ssl.CERT_OPTIONAL sock = ssl.wrap_socket(sock, keyfile=self.interface[2], certfile=self.interface[3], server_side=True, cert_reqs=cert_reqs, ca_certs=ca_certs, ssl_version=ssl.PROTOCOL_SSLv23) else: sock = ssl.wrap_socket(sock, keyfile=self.interface[2], certfile=self.interface[3], server_side=True, ssl_version=ssl.PROTOCOL_SSLv23) except SSLError: # Generally this happens when an HTTP request is received on a # secure socket. We don't do anything because it will be detected # by Worker and dealt with appropriately. pass return sock
def send (self, data): if self._closed: # usually handshaking failure, already handled exception return try: numsent = self.socket.send (data) if numsent: self.set_event_time () return numsent except ssl.SSLError as why: if why.errno == ssl.SSL_ERROR_WANT_WRITE: return 0 elif why.errno == ssl.SSL_ERROR_ZERO_RETURN: self.handle_close (700, "Connection closed by SSL_ERROR_ZERO_RETURN") return 0 else: raise
def handle_connect_event(self): if not self.handshaking: err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) if err != 0: raise socket.error(err, _strerror(err)) self.socket = ssl.wrap_socket (self.socket, do_handshake_on_connect = False) self.handshaking = True try: self.socket.do_handshake () except ssl.SSLError as why: if why.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): return # retry handshake raise ssl.SSLError(why) # handshaking done self.handle_connect() self.connected = True
def recv (self, buffer_size): self.event_time = time.time () try: data = self.socket.recv (buffer_size) if not data: self.handle_close () return b'' else: return data except ssl.SSLError as why: if why.errno == ssl.SSL_ERROR_WANT_READ: return b'' # retry # closed connection elif why.errno == ssl.SSL_ERROR_EOF: self.log ("SSL_ERROR_EOF Error Occurred in recv ()", "warn") self.handle_close () return b'' else: raise
def testInternalRecvStrict(self): sock = ws.WebSocket() s = sock.sock = SockMock() s.add_packet(six.b("foo")) s.add_packet(socket.timeout()) s.add_packet(six.b("bar")) # s.add_packet(SSLError("The read operation timed out")) s.add_packet(six.b("baz")) with self.assertRaises(ws.WebSocketTimeoutException): data = sock.frame_buffer.recv_strict(9) # if six.PY2: # with self.assertRaises(ws.WebSocketTimeoutException): # data = sock._recv_strict(9) # else: # with self.assertRaises(SSLError): # data = sock._recv_strict(9) data = sock.frame_buffer.recv_strict(9) self.assertEqual(data, six.b("foobarbaz")) with self.assertRaises(ws.WebSocketConnectionClosedException): data = sock.frame_buffer.recv_strict(1)
def _cert_array_from_pem(pem_bundle): """ Given a bundle of certs in PEM format, turns them into a CFArray of certs that can be used to validate a cert chain. """ der_certs = [ base64.b64decode(match.group(1)) for match in _PEM_CERTS_RE.finditer(pem_bundle) ] if not der_certs: raise ssl.SSLError("No root certificates specified") cert_array = CoreFoundation.CFArrayCreateMutable( CoreFoundation.kCFAllocatorDefault, 0, ctypes.byref(CoreFoundation.kCFTypeArrayCallBacks) ) if not cert_array: raise ssl.SSLError("Unable to allocate memory!") try: for der_bytes in der_certs: certdata = _cf_data_from_bytes(der_bytes) if not certdata: raise ssl.SSLError("Unable to allocate memory!") cert = Security.SecCertificateCreateWithData( CoreFoundation.kCFAllocatorDefault, certdata ) CoreFoundation.CFRelease(certdata) if not cert: raise ssl.SSLError("Unable to build cert object!") CoreFoundation.CFArrayAppendValue(cert_array, cert) CoreFoundation.CFRelease(cert) except Exception: # We need to free the array before the exception bubbles further. # We only want to do that if an error occurs: otherwise, the caller # should free. CoreFoundation.CFRelease(cert_array) return cert_array
def wrap_socket(self, sock, server_side=False, do_handshake_on_connect=True, suppress_ragged_eofs=True, server_hostname=None): cnx = OpenSSL.SSL.Connection(self._ctx, sock) if isinstance(server_hostname, six.text_type): # Platform-specific: Python 3 server_hostname = server_hostname.encode('utf-8') if server_hostname is not None: cnx.set_tlsext_host_name(server_hostname) cnx.set_connect_state() while True: try: cnx.do_handshake() except OpenSSL.SSL.WantReadError: rd = util.wait_for_read(sock, sock.gettimeout()) if not rd: raise timeout('select timed out') continue except OpenSSL.SSL.Error as e: raise ssl.SSLError('bad handshake: %r' % e) break return WrappedSocket(cnx, sock)
def _retry_request(http, num_retries, req_type, sleep, rand, uri, method, *args, **kwargs): """Retries an HTTP request multiple times while handling errors. If after all retries the request still fails, last error is either returned as return value (for HTTP 5xx errors) or thrown (for ssl.SSLError). Args: http: Http object to be used to execute request. num_retries: Maximum number of retries. req_type: Type of the request (used for logging retries). sleep, rand: Functions to sleep for random time between retries. uri: URI to be requested. method: HTTP method to be used. args, kwargs: Additional arguments passed to http.request. Returns: resp, content - Response from the http request (may be HTTP 5xx). """ resp = None for retry_num in range(num_retries + 1): if retry_num > 0: sleep(rand() * 2**retry_num) logging.warning( 'Retry #%d for %s: %s %s%s' % (retry_num, req_type, method, uri, ', following status: %d' % resp.status if resp else '')) try: resp, content = http.request(uri, method, *args, **kwargs) except ssl.SSLError: if retry_num == num_retries: raise else: continue if resp.status < 500: break return resp, content
def handle_read(self, watcher, revents, errno=None): if revents & libev.EV_ERROR: if errno: exc = IOError(errno, os.strerror(errno)) else: exc = Exception("libev reported an error") self.defunct(exc) return try: while True: buf = self._socket.recv(self.in_buffer_size) self._iobuf.write(buf) if len(buf) < self.in_buffer_size: break except socket.error as err: if ssl and isinstance(err, ssl.SSLError): if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): self.defunct(err) return elif err.args[0] not in NONBLOCKING: self.defunct(err) return if self._iobuf.tell(): self.process_io_buffer() else: log.debug("Connection %s closed by server", self) self.close()
def ssl_wrap_socket(sock, keyfile=None, certfile=None, cert_reqs=None, ca_certs=None, server_hostname=None, ssl_version=None): ctx = OpenSSL.SSL.Context(_openssl_versions[ssl_version]) if certfile: ctx.use_certificate_file(certfile) if keyfile: ctx.use_privatekey_file(keyfile) if cert_reqs != ssl.CERT_NONE: ctx.set_verify(_openssl_verify[cert_reqs], _verify_callback) if ca_certs: try: ctx.load_verify_locations(ca_certs, None) except OpenSSL.SSL.Error as e: raise ssl.SSLError('bad ca_certs: %r' % ca_certs, e) # Disable TLS compression to migitate CRIME attack (issue #309) OP_NO_COMPRESSION = 0x20000 ctx.set_options(OP_NO_COMPRESSION) # Set list of supported ciphersuites. ctx.set_cipher_list(DEFAULT_SSL_CIPHER_LIST) cnx = OpenSSL.SSL.Connection(ctx, sock) cnx.set_tlsext_host_name(server_hostname) cnx.set_connect_state() while True: try: cnx.do_handshake() except OpenSSL.SSL.WantReadError: select.select([sock], [], []) continue except OpenSSL.SSL.Error as e: raise ssl.SSLError('bad handshake', e) break return WrappedSocket(cnx, sock)
def is_ssl_error(error=None): """Checks if the given error (or the current one) is an SSL error.""" exc_types = (ssl.SSLError,) try: from OpenSSL.SSL import Error exc_types += (Error,) except ImportError: pass if error is None: error = sys.exc_info()[1] return isinstance(error, exc_types)
def _tls_match_hostname(self): cert = self._ssl.getpeercert() san = cert.get('subjectAltName') if san: have_san_dns = False for (key, value) in san: if key == 'DNS': have_san_dns = True if self._host_matches_cert(self._host.lower(), value.lower()) == True: return if key == 'IP Address': have_san_dns = True if value.lower() == self._host.lower(): return if have_san_dns: # Only check subject if subjectAltName dns not found. raise ssl.SSLError('Certificate subject does not match remote hostname.') subject = cert.get('subject') if subject: for ((key, value),) in subject: if key == 'commonName': if self._host_matches_cert(self._host.lower(), value.lower()) == True: return raise ssl.SSLError('Certificate subject does not match remote hostname.') # Compatibility class for easy porting from mosquitto.py.
