Python twisted.internet.endpoints 模块，SSL4ClientEndpoint() 实例源码

def _getEndpoint(self, scheme, host, port):
        """
        Get an endpoint for the given host and port, using a transport
        selected based on scheme.

        @param scheme: A string like C{'http'} or C{'https'} (the only two
            supported values) to use to determine how to establish the
            connection.

        @param host: A C{str} giving the hostname which will be connected to in
            order to issue a request.

        @param port: An C{int} giving the port number the connection will be
            on.

        @return: An endpoint which can be used to connect to given address.
        """
        kwargs = {}
        if self._connectTimeout is not None:
            kwargs['timeout'] = self._connectTimeout
        kwargs['bindAddress'] = self._bindAddress
        if scheme == 'http':
            return TCP4ClientEndpoint(self._reactor, host, port, **kwargs)
        elif scheme == 'https':
            return SSL4ClientEndpoint(self._reactor, host, port,
                                      self._wrapContextFactory(host, port),
                                      **kwargs)
        else:
            raise SchemeNotSupported("Unsupported scheme: %r" % (scheme,))

def test_sslPositionalArgs(self):
        """
        When passed an SSL strports description, L{clientFromString} returns a
        L{SSL4ClientEndpoint} instance initialized with the values from the
        string.
        """
        reactor = object()
        client = endpoints.clientFromString(
            reactor,
            "ssl:example.net:4321:privateKey=%s:"
            "certKey=%s:bindAddress=10.0.0.3:timeout=3:caCertsDir=%s" %
            (escapedPEMPathName, escapedPEMPathName, escapedCAsPathName))
        self.assertIsInstance(client, endpoints.SSL4ClientEndpoint)
        self.assertIs(client._reactor, reactor)
        self.assertEqual(client._host, "example.net")
        self.assertEqual(client._port, 4321)
        self.assertEqual(client._timeout, 3)
        self.assertEqual(client._bindAddress, ("10.0.0.3", 0))

def test_sslWithDefaults(self):
        """
        When passed an SSL strports description without extra arguments,
        L{clientFromString} returns a L{SSL4ClientEndpoint} instance
        whose context factory is initialized with default values.
        """
        reactor = object()
        client = endpoints.clientFromString(reactor, "ssl:example.net:4321")
        self.assertIsInstance(client, endpoints.SSL4ClientEndpoint)
        self.assertIs(client._reactor, reactor)
        self.assertEqual(client._host, "example.net")
        self.assertEqual(client._port, 4321)
        certOptions = client._sslContextFactory
        self.assertEqual(certOptions.method, SSLv23_METHOD)
        self.assertIsNone(certOptions.certificate)
        self.assertIsNone(certOptions.privateKey)

def main(reactor):
    pem = generate_certs.and_generate()
    caPem = FilePath(b"ca-private-cert.pem").getContent()
    clientEndpoint = SSL4ClientEndpoint(
        reactor, u"localhost", 4321,
        optionsForClientTLS(u"the-authority", Certificate.loadPEM(caPem),
                            PrivateCertificate.loadPEM(pem)),
    )
    clientEndpoint = SSL4ClientEndpoint(
        reactor, u"localhost", 4321,
        optionsForClientTLS(u"the-authority", Certificate.loadPEM(caPem),
                            PrivateCertificate.loadPEM(pem)),
    )
    proto = yield clientEndpoint.connect(Factory.forProtocol(SendAnyData))
    yield proto.deferred

def subEndpoint(self, reactor, host, port, contextFactory):
        """
        Create an endpoint to connect to based on a single address result from
        L{getaddrinfo}.

        @param reactor: the reactor to connect to
        @type reactor: L{IReactorTCP}

        @param host: The IP address of the host to connect to, in presentation
            format.
        @type host: L{str}

        @param port: The numeric port number to connect to.
        @type port: L{int}

        @param contextFactory: If not L{None}, the OpenSSL context factory to
            use to produce client connections.

        @return: a stream client endpoint that will connect to the given host
            and port via the given reactor.
        @rtype: L{IStreamClientEndpoint}
        """
        if contextFactory is None:
            return TCP4ClientEndpoint(reactor, host, port)
        else:
            return SSL4ClientEndpoint(reactor, host, port, contextFactory)

def createClientEndpoint(self, reactor, clientFactory, **connectArgs):
        """
        Create an L{SSL4ClientEndpoint} and return the values needed to verify
        its behaviour.

        @param reactor: A fake L{IReactorSSL} that L{SSL4ClientEndpoint} can
            call L{IReactorSSL.connectSSL} on.
        @param clientFactory: The thing that we expect to be passed to our
            L{IStreamClientEndpoint.connect} implementation.
        @param connectArgs: Optional dictionary of arguments to
            L{IReactorSSL.connectSSL}
        """
        address = IPv4Address("TCP", "localhost", 80)

        if connectArgs is None:
            connectArgs = {}

        return (endpoints.SSL4ClientEndpoint(reactor,
                                             address.host,
                                             address.port,
                                             self.clientSSLContext,
                                             **connectArgs),
                (address.host, address.port, clientFactory,
                 self.clientSSLContext,
                 connectArgs.get('timeout', 30),
                 connectArgs.get('bindAddress', None)),
                address)

def __init__(self,
                 url,
                 realm=None,
                 extra=None,
                 serializers=None,
                 ssl=None,
                 proxy=None,
                 headers=None):
        """

        :param url: The WebSocket URL of the WAMP router to connect to (e.g. `ws://somehost.com:8090/somepath`)
        :type url: str

        :param realm: The WAMP realm to join the application session to.
        :type realm: str

        :param extra: Optional extra configuration to forward to the application component.
        :type extra: dict

        :param serializers: A list of WAMP serializers to use (or None for default serializers).
           Serializers must implement :class:`autobahn.wamp.interfaces.ISerializer`.
        :type serializers: list

        :param ssl: (Optional). If specified this should be an
            instance suitable to pass as ``sslContextFactory`` to
            :class:`twisted.internet.endpoints.SSL4ClientEndpoint`` such
            as :class:`twisted.internet.ssl.CertificateOptions`. Leaving
            it as ``None`` will use the result of calling Twisted's
            :meth:`twisted.internet.ssl.platformTrust` which tries to use
            your distribution's CA certificates.
        :type ssl: :class:`twisted.internet.ssl.CertificateOptions`

        :param proxy: Explicit proxy server to use; a dict with ``host`` and ``port`` keys
        :type proxy: dict or None

        :param headers: Additional headers to send (only applies to WAMP-over-WebSocket).
        :type headers: dict
        """
        assert(type(url) == six.text_type)
        assert(realm is None or type(realm) == six.text_type)
        assert(extra is None or type(extra) == dict)
        assert(headers is None or type(headers) == dict)
        assert(proxy is None or type(proxy) == dict)
        self.url = url
        self.realm = realm
        self.extra = extra or dict()
        self.serializers = serializers
        self.ssl = ssl
        self.proxy = proxy
        self.headers = headers

        # this if for auto-reconnection when Twisted ClientService is avail
        self._client_service = None
        # total number of successful connections
        self._connect_successes = 0

def test_ssl(self):
        """
        When passed an SSL strports description, L{clientFromString} returns a
        L{SSL4ClientEndpoint} instance initialized with the values from the
        string.
        """
        reactor = object()
        client = endpoints.clientFromString(
            reactor,
            "ssl:host=example.net:port=4321:privateKey=%s:"
            "certKey=%s:bindAddress=10.0.0.3:timeout=3:caCertsDir=%s" %
            (escapedPEMPathName, escapedPEMPathName, escapedCAsPathName))
        self.assertIsInstance(client, endpoints.SSL4ClientEndpoint)
        self.assertIs(client._reactor, reactor)
        self.assertEqual(client._host, "example.net")
        self.assertEqual(client._port, 4321)
        self.assertEqual(client._timeout, 3)
        self.assertEqual(client._bindAddress, ("10.0.0.3", 0))
        certOptions = client._sslContextFactory
        self.assertIsInstance(certOptions, CertificateOptions)
        self.assertEqual(certOptions.method, SSLv23_METHOD)
        self.assertTrue(certOptions._options & OP_NO_SSLv3)
        ctx = certOptions.getContext()
        self.assertIsInstance(ctx, ContextType)
        self.assertEqual(Certificate(certOptions.certificate), testCertificate)
        privateCert = PrivateCertificate(certOptions.certificate)
        privateCert._setPrivateKey(KeyPair(certOptions.privateKey))
        self.assertEqual(privateCert, testPrivateCertificate)
        expectedCerts = [
            Certificate.loadPEM(x.getContent()) for x in
            [casPath.child("thing1.pem"), casPath.child("thing2.pem")]
            if x.basename().lower().endswith('.pem')
        ]
        addedCerts = []
        class ListCtx(object):
            def get_cert_store(self):
                class Store(object):
                    def add_cert(self, cert):
                        addedCerts.append(cert)
                return Store()
        certOptions.trustRoot._addCACertsToContext(ListCtx())
        self.assertEqual(
            sorted((Certificate(x) for x in addedCerts),
                   key=lambda cert: cert.digest()),
            sorted(expectedCerts,
                   key=lambda cert: cert.digest())
        )