我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用ssl.wrap_socket()。
def connect(self): sock = socket.create_connection( (self.host, self.port), getattr(self, 'source_address', None) ) # Handle the socket if a (proxy) tunnel is present if hasattr(self, '_tunnel') and getattr(self, '_tunnel_host', None): self.sock = sock self._tunnel() self.sock = ssl.wrap_socket( sock, cert_reqs=ssl.CERT_REQUIRED, ca_certs=self.ca_bundle ) try: match_hostname(self.sock.getpeercert(), self.host) except CertificateError: self.sock.shutdown(socket.SHUT_RDWR) self.sock.close() raise
def loop(self): """Main server loop for accepting connections. Better call it on its own thread""" while True: try: (csock, (ipaddr, port)) = self.connection["sock"].accept() self._log("L", "New connection from %s:%s" % (str(ipaddr), str(port))) except sock_error: raise sock_error try: csock = ssl.wrap_socket(csock, server_side=True, certfile="server.crt", keyfile="server.key", ssl_version=ssl.PROTOCOL_TLSv1_2) except AttributeError: # All PROTOCOL consts are merged on TLS in Python2.7.13 csock = ssl.wrap_socket(csock, server_side=True, certfile="server.crt", keyfile="server.key", ssl_version=ssl.PROTOCOL_TLS) self.clients["hosts"][str(self.clients["serial"])] = Host(csock, ipaddr, port, self.clients["serial"]) self.clients["serial"] += 1
def wrap_socket(self, socket, server_hostname=None, server_side=False): warnings.warn( 'A true SSLContext object is not available. This prevents ' 'urllib3 from configuring SSL appropriately and may cause ' 'certain SSL connections to fail. You can upgrade to a newer ' 'version of Python to solve this. For more information, see ' 'https://urllib3.readthedocs.io/en/latest/security.html' '#insecureplatformwarning.', InsecurePlatformWarning ) kwargs = { 'keyfile': self.keyfile, 'certfile': self.certfile, 'ca_certs': self.ca_certs, 'cert_reqs': self.verify_mode, 'ssl_version': self.protocol, 'server_side': server_side, } if self.supports_set_ciphers: # Platform-specific: Python 2.7+ return wrap_socket(socket, ciphers=self.ciphers, **kwargs) else: # Platform-specific: Python 2.6 return wrap_socket(socket, **kwargs)
def resolve_cert_reqs(candidate): """ Resolves the argument to a numeric constant, which can be passed to the wrap_socket function/method from the ssl module. Defaults to :data:`ssl.CERT_NONE`. If given a string it is assumed to be the name of the constant in the :mod:`ssl` module or its abbrevation. (So you can specify `REQUIRED` instead of `CERT_REQUIRED`. If it's neither `None` nor a string we assume it is already the numeric constant which can directly be passed to wrap_socket. """ if candidate is None: return CERT_NONE if isinstance(candidate, str): res = getattr(ssl, candidate, None) if res is None: res = getattr(ssl, 'CERT_' + candidate) return res return candidate
def start(self): self.deviceHandler.start() if self.protocol == "udp": self.loadState() self.logger.debug("udpHeartbeatSeconds = {0}".format(self.udpHeartbeatSeconds)) self.logger.debug("udpDataPacketInterval = {0}".format(self.udpDataPacketInterval)) self.udpServer = SocketServer.UDPServer(('0.0.0.0', 0), IotUDPHandler) self.udpServer.service = self self.udpServer.role = IotUDPHandler.CLIENT self.logger.info("starting UDP client at {0}:{1} connecting to {2}, state at {3}".format(self.udpServer.server_address[0], self.udpServer.server_address[1], self.serverAddr, self.stateFile)) timer = threading.Timer(0.5, self.repeat) timer.daemon = True timer.start() self.udpServer.serve_forever() elif self.protocol == "ssl": while True: self.logger.info("Connecting by SSL to server at {0}".format(self.serverAddr)) try: sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM) self.logger.debug("using caCertFile={0}, deviceCertFile={1}, deviceKeyFile={2}".format(self.caCertFile, self.deviceCertFile, self.deviceKeyFile)) sslSocket = ssl.wrap_socket(sock, ca_certs=self.caCertFile, cert_reqs=ssl.CERT_REQUIRED, certfile=self.deviceCertFile, keyfile=self.deviceKeyFile, ssl_version=ssl.PROTOCOL_TLSv1) sslSocket.connect((self.serverAddr.split(':')[0], int(self.serverAddr.split(':')[1]))) servercert = sslSocket.getpeercert() subject = dict(x[0] for x in servercert['subject']) self.logger.info("Connected to server with valid certificate, CN={0}".format(subject['commonName'])) self.sslSocket = sslSocket sslThread = threading.Thread(target = self.sslListen, args = (self.sslSocket,)) sslThread.daemon = True sslThread.start() while True: payload = self.deviceHandler.getMessagePayload() self.logger.debug("Sending payload to {0} by SSL: {1}".format(self.serverAddr, payload)) iotcommon.sendMessage(self.sslSocket, payload) time.sleep(self.sslIntervalSeconds) except Exception as e: self.logger.exception(e) time.sleep(10)
def wrap_socket(self, socket, server_hostname=None, server_side=False): warnings.warn( 'A true SSLContext object is not available. This prevents ' 'urllib3 from configuring SSL appropriately and may cause ' 'certain SSL connections to fail. You can upgrade to a newer ' 'version of Python to solve this. For more information, see ' 'https://urllib3.readthedocs.io/en/latest/advanced-usage.html' '#ssl-warnings', InsecurePlatformWarning ) kwargs = { 'keyfile': self.keyfile, 'certfile': self.certfile, 'ca_certs': self.ca_certs, 'cert_reqs': self.verify_mode, 'ssl_version': self.protocol, 'server_side': server_side, } if self.supports_set_ciphers: # Platform-specific: Python 2.7+ return wrap_socket(socket, ciphers=self.ciphers, **kwargs) else: # Platform-specific: Python 2.6 return wrap_socket(socket, **kwargs)
def main(args): global HTTPD, CREDENTIALS if args: load_settings(args[0]) print("Starting server") server_address = (LISTENIP, LISTENPORT) if CREDENTIALS: CREDENTIALS = base64.b64encode(bytes(CREDENTIALS, "utf-8")) Handler = AuthHandler else: Handler = RequestHandler if not SSL_CERTIFICATE: HTTPD = HTTPServer(server_address, Handler) else: HTTPD = socketserver.TCPServer(server_address, Handler) HTTPD.socket = ssl.wrap_socket(HTTPD.socket, certfile=SSL_CERTIFICATE, keyfile=SSL_KEY, server_side=True) print('Listening on: %s://%s:%i' % ('https' if SSL_CERTIFICATE else 'http', LISTENIP, LISTENPORT)) if BASEPATH: os.chdir(BASEPATH) HTTPD.serve_forever()
def send(self,send_data): if self.__is_addr_reset or \ self._is_close_for_invoker or \ self.__is_connection_closed(): self.__connection = socket(AF_INET,SOCK_STREAM) if self.__is_ssl: self.__connection = ssl.wrap_socket(self.__connection) try : self.__connection.connect(self.__addr) self.__connection.settimeout(self.__timeout) self._is_close_for_invoker = False self.__is_addr_reset = False except Exception as exception: self.__recorder.write( 'Cannot connect %s' % str(self.__addr), Exception = str(exception)) return None self.__connection.send(send_data) return self
def __init__(self, host, port = POP3_SSL_PORT, keyfile = None, certfile = None): self.host = host self.port = port self.keyfile = keyfile self.certfile = certfile self.buffer = "" msg = "getaddrinfo returns an empty list" self.sock = None for res in socket.getaddrinfo(self.host, self.port, 0, socket.SOCK_STREAM): af, socktype, proto, canonname, sa = res try: self.sock = socket.socket(af, socktype, proto) self.sock.connect(sa) except socket.error, msg: if self.sock: self.sock.close() self.sock = None continue break if not self.sock: raise socket.error, msg self.file = self.sock.makefile('rb') self.sslobj = ssl.wrap_socket(self.sock, self.keyfile, self.certfile) self._debugging = 0 self.welcome = self._getresp()
def wrap_socket(self, socket, server_hostname=None, server_side=False): warnings.warn( 'A true SSLContext object is not available. This prevents ' 'urllib3 from configuring SSL appropriately and may cause ' 'certain SSL connections to fail. You can upgrade to a newer ' 'version of Python to solve this. For more information, see ' 'https://urllib3.readthedocs.org/en/latest/security.html' '#insecureplatformwarning.', InsecurePlatformWarning ) kwargs = { 'keyfile': self.keyfile, 'certfile': self.certfile, 'ca_certs': self.ca_certs, 'cert_reqs': self.verify_mode, 'ssl_version': self.protocol, 'server_side': server_side, } if self.supports_set_ciphers: # Platform-specific: Python 2.7+ return wrap_socket(socket, ciphers=self.ciphers, **kwargs) else: # Platform-specific: Python 2.6 return wrap_socket(socket, **kwargs)
def __init__(self, dispatcher, connection, address, use_ssl, ssl_certfile, ssl_keyfile): Session.__init__(self, dispatcher) self.use_ssl = use_ssl self.raw_connection = connection if use_ssl: import ssl self._connection = ssl.wrap_socket( connection, server_side=True, certfile=ssl_certfile, keyfile=ssl_keyfile, ssl_version=ssl.PROTOCOL_SSLv23, do_handshake_on_connect=False) else: self._connection = connection self.address = address[0] + ":%d"%address[1] self.name = "TCP " if not use_ssl else "SSL " self.timeout = 1000 self.dispatcher.add_session(self) self.response_queue = queue.Queue() self.message = '' self.retry_msg = '' self.handshake = not self.use_ssl self.need_write = True
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 __init__(self, *args, **kwargs): """The ``ssl_options`` keyword argument may either be an `ssl.SSLContext` object or a dictionary of keywords arguments for `ssl.wrap_socket` """ self._ssl_options = kwargs.pop('ssl_options', _client_ssl_defaults) super(SSLIOStream, self).__init__(*args, **kwargs) self._ssl_accepting = True self._handshake_reading = False self._handshake_writing = False self._ssl_connect_callback = None self._server_hostname = None # If the socket is already connected, attempt to start the handshake. try: self.socket.getpeername() except socket.error: pass else: # Indirectly start the handshake, which will run on the next # IOLoop iteration and then the real IO state will be set in # _handle_events. self._add_io_state(self.io_loop.WRITE)
def ssl_wrap_socket(socket, ssl_options, server_hostname=None, **kwargs): """Returns an ``ssl.SSLSocket`` wrapping the given socket. ``ssl_options`` may be either an `ssl.SSLContext` object or a dictionary (as accepted by `ssl_options_to_context`). Additional keyword arguments are passed to ``wrap_socket`` (either the `~ssl.SSLContext` method or the `ssl` module function as appropriate). """ context = ssl_options_to_context(ssl_options) if hasattr(ssl, 'SSLContext') and isinstance(context, ssl.SSLContext): if server_hostname is not None and getattr(ssl, 'HAS_SNI'): # Python doesn't have server-side SNI support so we can't # really unittest this, but it can be manually tested with # python3.2 -m tornado.httpclient https://sni.velox.ch return context.wrap_socket(socket, server_hostname=server_hostname, **kwargs) else: return context.wrap_socket(socket, **kwargs) else: return ssl.wrap_socket(socket, **dict(context, **kwargs))
def _handle_connect(self): # Call the superclass method to check for errors. super(SSLIOStream, self)._handle_connect() if self.closed(): return # When the connection is complete, wrap the socket for SSL # traffic. Note that we do this by overriding _handle_connect # instead of by passing a callback to super().connect because # user callbacks are enqueued asynchronously on the IOLoop, # but since _handle_events calls _handle_connect immediately # followed by _handle_write we need this to be synchronous. # # The IOLoop will get confused if we swap out self.socket while the # fd is registered, so remove it now and re-register after # wrap_socket(). self.io_loop.remove_handler(self.socket) old_state = self._state self._state = None self.socket = ssl_wrap_socket(self.socket, self._ssl_options, server_hostname=self._server_hostname, do_handshake_on_connect=False) self._add_io_state(old_state)
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 open(self, host="", port=_DEFAULT_PORT_IMAP4_SSL): self.host = host self.port = port self.sock = socket.create_connection((host, port), timeout=self._timeout) with ca_certs(self.ca_certs) as certs: self.sslobj = ssl.wrap_socket( self.sock, keyfile=self.keyfile, certfile=self.certfile, cert_reqs=ssl.CERT_REQUIRED, ca_certs=certs ) cert = self.sslobj.getpeercert() match_hostname(cert, host) self.file = self.sslobj.makefile("rb")
def connect(self): httplib.HTTPConnection.connect(self) with ca_certs(self.ca_certs) as certs: self.sock = ssl.wrap_socket( self.sock, certfile=self.certfile, keyfile=self.keyfile, cert_reqs=ssl.CERT_REQUIRED if self.require_cert else ssl.CERT_NONE, ca_certs=certs ) if self.require_cert: hostname = self.host if not self._tunnel_host else self._tunnel_host cert = self.sock.getpeercert() match_hostname(cert, hostname)
