我们从Python开源项目中，提取了以下6个代码示例，用于说明如何使用gevent.ssl()。
def openLibrary(): import ctypes import ctypes.util try: if sys.platform.startswith("win"): dll_path = "src/lib/opensslVerify/libeay32.dll" elif sys.platform == "cygwin": dll_path = "/bin/cygcrypto-1.0.0.dll" else: dll_path = "/usr/local/ssl/lib/libcrypto.so" ssl = ctypes.CDLL(dll_path, ctypes.RTLD_GLOBAL) assert ssl except: dll_path = ctypes.util.find_library('ssl') or ctypes.util.find_library('crypto') or ctypes.util.find_library('libcrypto') ssl = ctypes.CDLL(dll_path or 'libeay32', ctypes.RTLD_GLOBAL) return ssl
def ssl(sock, keyfile=None, certfile=None): # deprecated in 2.7.9 but still present; # sometimes backported by distros. See ssl.py from gevent import ssl as _sslmod # wrap_socket is 2.7.9/backport, sslwrap_simple is older. They take # the same arguments. wrap = getattr(_sslmod, 'wrap_socket', None) or getattr(_sslmod, 'sslwrap_simple') return wrap(sock, keyfile, certfile)
def ssl(sock, keyfile=None, certfile=None): # deprecated in 2.7.9 but still present; # sometimes backported by distros. See ssl.py # Note that we import gevent.ssl, not _ssl2, to get the correct # version. from gevent import ssl as _sslmod # wrap_socket is 2.7.9/backport, sslwrap_simple is older. They take # the same arguments. wrap = getattr(_sslmod, 'wrap_socket', None) or getattr(_sslmod, 'sslwrap_simple') return wrap(sock, keyfile, certfile)
def _connect( self ): try: self._socket = gevent.ssl.wrap_socket( gevent.socket.socket( gevent.socket.AF_INET, gevent.socket.SOCK_STREAM ), cert_reqs = gevent.ssl.CERT_NONE ) self._socket.connect( ( self._destServer, self._destPort ) ) self._log( "Connected" ) headers = rSequence() headers.addSequence( Symbols.base.HCP_IDENT, AgentId( ( self._oid, self._iid, self._sid, self._plat, self._arch ) ).toJson() ) headers.addStringA( Symbols.base.HOST_NAME, hashlib.md5( str( self._sid ) ).hexdigest() ) headers.addIpv4( Symbols.base.IP_ADDRESS, "%d.%d.%d.%d" % ( random.randint( 0, 254 ), random.randint( 0, 254 ), random.randint( 0, 254 ), random.randint( 0, 254 ) ) ) if self._enrollmentToken is not None: headers.addBuffer( Symbols.hcp.ENROLLMENT_TOKEN, self._enrollmentToken ) self._sendFrame( HcpModuleId.HCP, [ headers ], timeout = 30, isNotHbs = True ) self._log( "Handshake sent" ) self._threads.add( gevent.spawn( self._recvThread ) ) self._threads.add( gevent.spawn_later( 1, self._syncHcpThread ) ) self._threads.add( gevent.spawn_later( 10, self._syncHbsThread ) ) self._threads.add( gevent.spawn_later( 2, lambda: self._connectedEvent.set() ) ) return True except: self._log( "Failed to connect over TLS: %s" % traceback.format_exc() ) return False
