def _openEndpoint(self, endpoint, factory):
     if IStreamClientEndpoint.providedBy(endpoint):
         d = endpoint.connect(factory)
     elif IStreamServerEndpoint.providedBy(endpoint):
         d = endpoint.listen(factory)
     else:
         raise ValueError('must provide either IStreamClientEndpoint or IStreamServerEndpoint')

 def _connect(self, reactor, update_status):
     maker = self._tor_control_endpoint_maker
     with add_context(update_status, "making Tor control endpoint"):
         tor_control_endpoint = yield maker(reactor, update_status)
     assert IStreamClientEndpoint.providedBy(tor_control_endpoint)
     with add_context(update_status, "connecting to Tor"):
         tproto = yield txtorcon.build_tor_connection(tor_control_endpoint,
                                                      build_state=False)
     with add_context(update_status, "waiting for Tor bootstrap"):
         config = yield txtorcon.TorConfig.from_protocol(tproto)
     ports = list(config.SocksPort)
     # I've seen "9050", and "unix:/var/run/tor/socks WorldWritable"
     for port in ports:
         pieces = port.split()
         p = pieces[0]
         if p == txtorcon.DEFAULT_VALUE:
             p = "9050"
         try:
             portnum = int(p)
             socks_desc = "tcp:127.0.0.1:%d" % portnum
             self._socks_desc = socks_desc # stash for tests
             socks_endpoint = clientFromString(reactor, socks_desc)
             returnValue(socks_endpoint)
         except ValueError:
             pass
     raise ValueError("could not use config.SocksPort: %r" % (ports,))

    def web_agent(self, pool=None, socks_endpoint=None):
        """
        :param socks_endpoint: If ``None`` (the default), a suitable
            SOCKS port is chosen from our config (or added). If supplied,
            should be a Deferred which fires an IStreamClientEndpoint
            (e.g. the return-value from
            :meth:`txtorcon.TorConfig.socks_endpoint`) or an immediate
            IStreamClientEndpoint You probably don't need to mess with
            this.

        :param pool: passed on to the Agent (as ``pool=``)
        """
        # local import since not all platforms have this
        from txtorcon import web

        if socks_endpoint is None:
            socks_endpoint = _create_socks_endpoint(self._reactor, self._protocol)
        if not isinstance(socks_endpoint, Deferred):
            if not IStreamClientEndpoint.providedBy(socks_endpoint):
                raise ValueError(
                    "'socks_endpoint' should be a Deferred or an IStreamClient"
                    "Endpoint (got '{}')".format(type(socks_endpoint))
                )
        return web.tor_agent(
            self._reactor,
            socks_endpoint,
            pool=pool,
        )

 def test_socksport_endpoint_otherhost(self):
     n = FakeNode(BASECONFIG+"[tor]\nsocks.port = tcp:otherhost:1234\n")
     h1 = mock.Mock()
     with mock.patch("foolscap.connections.tor.socks_endpoint",
                     return_value=h1) as f:
         h = n._make_tor_handler()
         self.assertTrue(IStreamClientEndpoint.providedBy(f.mock_calls[0]))
         self.assertIdentical(h, h1)

 def test_socksport_unix_endpoint(self):
     n = FakeNode(BASECONFIG+"[tor]\nsocks.port = unix:/var/lib/fw-daemon/tor_socks.socket\n")
     h1 = mock.Mock()
     with mock.patch("foolscap.connections.tor.socks_endpoint",
                     return_value=h1) as f:
         h = n._make_tor_handler()
         self.assertTrue(IStreamClientEndpoint.providedBy(f.mock_calls[0]))
         self.assertIdentical(h, h1)

def control_endpoint(tor_control_endpoint):
    """Return a handler which connects to a pre-existing Tor process on the
    given control port.
    - tor_control_endpoint: a ClientEndpoint which points at the Tor control
      port
    """
    assert IStreamClientEndpoint.providedBy(tor_control_endpoint)
    return _ConnectedTor(lambda reactor, update_status: tor_control_endpoint)

 def try_endpoint(control_ep):
     assert IStreamClientEndpoint.providedBy(control_ep)
     proto = yield control_ep.connect(
         TorProtocolFactory(
             password_function=password_function
         )
     )
     config = yield TorConfig.from_protocol(proto)
     tor = Tor(reactor, proto, _tor_config=config)
     returnValue(tor)

 def clean(self, value):
     from twisted.internet.endpoints import clientFromString
     from twisted.internet.interfaces import IStreamClientEndpoint
     from twisted.internet import reactor
     if IStreamClientEndpoint.providedBy(value):
         # We got an actual endpoint object, useful for testing.
         return value
     try:
         return clientFromString(reactor, value)
     except ValueError:
         self.raise_config_error('is not a valid client endpoint')

 def test_socksport_endpoint_otherhost(self):
     h1 = mock.Mock()
     with mock.patch("foolscap.connections.tor.socks_endpoint",
                     return_value=h1) as f:
         config = config_from_string(
             "no-basedir",
             "fake.port",
             BASECONFIG + "[tor]\nsocks.port = tcp:otherhost:1234\n",
         )
         tor_provider = create_tor_provider(reactor, config)
         h = tor_provider.get_tor_handler()
     self.assertTrue(IStreamClientEndpoint.providedBy(f.mock_calls[0]))
     self.assertIdentical(h, h1)

    def connect(self, factory):
        # further wrap the protocol if we're doing TLS.
        # "pray i do not wrap the protocol further".
        if self._tls:
            # XXX requires Twisted 14+
            from twisted.internet.ssl import optionsForClientTLS
            if self._tls is True:
                context = optionsForClientTLS(self._host)
            else:
                context = self._tls
            tls_factory = tls.TLSMemoryBIOFactory(context, True, factory)
            socks_factory = _TorSocksFactory(
                self._host, self._port, 'CONNECT', tls_factory,
            )
        else:
            socks_factory = _TorSocksFactory(
                self._host, self._port, 'CONNECT', factory,
            )

        self._socks_factory = socks_factory
        # forward our address (when we get it) to any listeners
        self._socks_factory._get_address().addBoth(self._when_address.fire)
        # XXX isn't this just maybeDeferred()
        if isinstance(self._proxy_ep, Deferred):
            proxy_ep = yield self._proxy_ep
            if not IStreamClientEndpoint.providedBy(proxy_ep):
                raise ValueError(
                    "The Deferred provided as 'socks_endpoint' must "
                    "resolve to an IStreamClientEndpoint provider (got "
                    "{})".format(type(proxy_ep).__name__)
                )
        else:
            proxy_ep = self._proxy_ep

        # socks_proto = yield proxy_ep.connect(socks_factory)
        proto = yield proxy_ep.connect(socks_factory)
        wrapped_proto = yield proto.when_done()
        if self._tls:
            returnValue(wrapped_proto.wrappedProtocol)
        else:
            returnValue(wrapped_proto)

 def _check_native_endpoint(self, endpoint):
     if IStreamClientEndpoint.providedBy(endpoint):
         pass
     elif isinstance(endpoint, dict):
         if u'tls' in endpoint:
             tls = endpoint[u'tls']
             if isinstance(tls, (dict, bool)):
                 pass
             elif IOpenSSLClientConnectionCreator.providedBy(tls):
                 pass
             elif isinstance(tls, CertificateOptions):
                 pass
             else:
                 raise ValueError(
                     "'tls' configuration must be a dict, CertificateOptions or"
                     " IOpenSSLClientConnectionCreator provider"
                 )
     else:
         raise ValueError(
             "'endpoint' configuration must be a dict or IStreamClientEndpoint"
             " provider"
         )

def build_tor_connection(connection, build_state=True, wait_for_proto=True,
                         password_function=lambda: None):
    """
    This is used to build a valid TorState (which has .protocol for
    the TorControlProtocol). For example::

        from twisted.internet import reactor
        from twisted.internet.endpoints import TCP4ClientEndpoint
        import txtorcon

        def example(state):
            print "Fully bootstrapped state:",state
            print "   with bootstrapped protocol:",state.protocol

        d = txtorcon.build_tor_connection(TCP4ClientEndpoint(reactor,
                                                             "localhost",
                                                             9051))
        d.addCallback(example)
        reactor.run()

    :param password_function:
        See :class:`txtorcon.TorControlProtocol`

    :param build_state:
        If True (the default) a TorState object will be
        built as well. If False, just a TorControlProtocol will be
        returned via the Deferred.

    :return:
        a Deferred that fires with a TorControlProtocol or, if you
        specified build_state=True, a TorState. In both cases, the
        object has finished bootstrapping
        (i.e. TorControlProtocol.post_bootstrap or
        TorState.post_bootstap has fired, as needed)
    """

    if IStreamClientEndpoint.providedBy(connection):
        endpoint = connection

    elif isinstance(connection, tuple):
        if len(connection) == 2:
            reactor, socket = connection
            if (os.path.exists(socket) and
                os.stat(socket).st_mode & (stat.S_IRGRP | stat.S_IRUSR |
                                           stat.S_IROTH)):
                endpoint = UNIXClientEndpoint(reactor, socket)
            else:
                raise ValueError('Can\'t use "%s" as a socket' % (socket, ))
        elif len(connection) == 3:
            endpoint = TCP4ClientEndpoint(*connection)
        else:
            raise TypeError('Expected either a (reactor, socket)- or a '
                            '(reactor, host, port)-tuple for argument '
                            '"connection", got %s' % (connection, ))
    else:
        raise TypeError('Expected a (reactor, socket)- or a (reactor, host, '
                        'port)-tuple or an object implementing IStreamClient'
                        'Endpoint for argument "connection", got %s' %
                        (connection, ))

    d = endpoint.connect(TorProtocolFactory(password_function=password_function))
    if build_state:
        d.addCallback(build_state if callable(build_state) else _build_state)
    elif wait_for_proto:
        d.addCallback(wait_for_proto if callable(wait_for_proto) else
                      _wait_for_proto)
    return d

 def __init__(self, sam_endpoint, **kwargs):
     assert IStreamClientEndpoint.providedBy(sam_endpoint)
     self._sam_endpoint = sam_endpoint
     self._kwargs = kwargs

def _create_transport_endpoint(reactor, endpoint_config):
    """
    Create a Twisted client endpoint for a WAMP-over-XXX transport.
    """
    if IStreamClientEndpoint.providedBy(endpoint_config):
        endpoint = IStreamClientEndpoint(endpoint_config)
    else:
        # create a connecting TCP socket
        if endpoint_config[u'type'] == u'tcp':

            version = endpoint_config.get(u'version', 4)
            if version not in [4, 6]:
                raise ValueError('invalid IP version {} in client endpoint configuration'.format(version))

            host = endpoint_config[u'host']
            if type(host) != six.text_type:
                raise ValueError('invalid type {} for host in client endpoint configuration'.format(type(host)))

            port = endpoint_config[u'port']
            if type(port) not in six.integer_types:
                raise ValueError('invalid type {} for port in client endpoint configuration'.format(type(port)))

            timeout = endpoint_config.get(u'timeout', 10)  # in seconds
            if type(timeout) not in six.integer_types:
                raise ValueError('invalid type {} for timeout in client endpoint configuration'.format(type(timeout)))

            tls = endpoint_config.get(u'tls', None)

            # create a TLS enabled connecting TCP socket
            if tls:
                if not _TLS:
                    raise RuntimeError('TLS configured in transport, but TLS support is not installed (eg OpenSSL?)')

                # FIXME: create TLS context from configuration
                if IOpenSSLClientConnectionCreator.providedBy(tls):
                    # eg created from twisted.internet.ssl.optionsForClientTLS()
                    context = IOpenSSLClientConnectionCreator(tls)

                elif isinstance(tls, dict):
                    for k in tls.keys():
                        if k not in [u"hostname", u"trust_root"]:
                            raise ValueError("Invalid key '{}' in 'tls' config".format(k))
                    hostname = tls.get(u'hostname', host)
                    if type(hostname) != six.text_type:
                        raise ValueError('invalid type {} for hostname in TLS client endpoint configuration'.format(hostname))
                    trust_root = None
                    cert_fname = tls.get(u"trust_root", None)
                    if cert_fname is not None:
                        trust_root = Certificate.loadPEM(six.u(open(cert_fname, 'r').read()))
                    context = optionsForClientTLS(hostname, trustRoot=trust_root)

                elif isinstance(tls, CertificateOptions):
                    context = tls

                elif tls is True:
                    context = optionsForClientTLS(host)

                else:
                    raise RuntimeError('unknown type {} for "tls" configuration in transport'.format(type(tls)))

                if version == 4:
                    endpoint = SSL4ClientEndpoint(reactor, host, port, context, timeout=timeout)
                elif version == 6:
                    # there is no SSL6ClientEndpoint!
                    raise RuntimeError('TLS on IPv6 not implemented')
                else:
                    assert(False), 'should not arrive here'

            # create a non-TLS connecting TCP socket
            else:
                if version == 4:
                    endpoint = TCP4ClientEndpoint(reactor, host, port, timeout=timeout)
                elif version == 6:
                    try:
                        from twisted.internet.endpoints import TCP6ClientEndpoint
                    except ImportError:
                        raise RuntimeError('IPv6 is not supported (please upgrade Twisted)')
                    endpoint = TCP6ClientEndpoint(reactor, host, port, timeout=timeout)
                else:
                    assert(False), 'should not arrive here'

        # create a connecting Unix domain socket
        elif endpoint_config[u'type'] == u'unix':
            path = endpoint_config[u'path']
            timeout = int(endpoint_config.get(u'timeout', 10))  # in seconds
            endpoint = UNIXClientEndpoint(reactor, path, timeout=timeout)

        else:
            assert(False), 'should not arrive here'

    return endpoint

def socks_endpoint(tor_socks_endpoint):
    assert IStreamClientEndpoint.providedBy(tor_socks_endpoint)
    return _SocksTor(tor_socks_endpoint)

def _create_transport_endpoint(reactor, endpoint_config):
    """
    Create a Twisted client endpoint for a WAMP-over-XXX transport.
    """
    if IStreamClientEndpoint.providedBy(endpoint_config):
        endpoint = IStreamClientEndpoint(endpoint_config)
    else:
        # create a connecting TCP socket
        if endpoint_config['type'] == 'tcp':

            version = int(endpoint_config.get('version', 4))
            host = str(endpoint_config['host'])
            port = int(endpoint_config['port'])
            timeout = int(endpoint_config.get('timeout', 10))  # in seconds
            tls = endpoint_config.get('tls', None)

            # create a TLS enabled connecting TCP socket
            if tls:
                if not _TLS:
                    raise RuntimeError('TLS configured in transport, but TLS support is not installed (eg OpenSSL?)')

                # FIXME: create TLS context from configuration
                if IOpenSSLClientConnectionCreator.providedBy(tls):
                    # eg created from twisted.internet.ssl.optionsForClientTLS()
                    context = IOpenSSLClientConnectionCreator(tls)

                elif isinstance(tls, CertificateOptions):
                    context = tls

                elif tls is True:
                    context = optionsForClientTLS(host)

                else:
                    raise RuntimeError('unknown type {} for "tls" configuration in transport'.format(type(tls)))

                if version == 4:
                    endpoint = SSL4ClientEndpoint(reactor, host, port, context, timeout=timeout)
                elif version == 6:
                    # there is no SSL6ClientEndpoint!
                    raise RuntimeError('TLS on IPv6 not implemented')
                else:
                    assert(False), 'should not arrive here'

            # create a non-TLS connecting TCP socket
            else:
                if version == 4:
                    endpoint = TCP4ClientEndpoint(reactor, host, port, timeout=timeout)
                elif version == 6:
                    try:
                        from twisted.internet.endpoints import TCP6ClientEndpoint
                    except ImportError:
                        raise RuntimeError('IPv6 is not supported (please upgrade Twisted)')
                    endpoint = TCP6ClientEndpoint(reactor, host, port, timeout=timeout)
                else:
                    assert(False), 'should not arrive here'

        # create a connecting Unix domain socket
        elif endpoint_config['type'] == 'unix':
            path = endpoint_config['path']
            timeout = int(endpoint_config.get('timeout', 10))  # in seconds
            endpoint = UNIXClientEndpoint(reactor, path, timeout=timeout)

        else:
            assert(False), 'should not arrive here'

    return endpoint

 def __init__(self, sam_endpoint):
     assert IStreamClientEndpoint.providedBy(sam_endpoint)
     self._sam_endpoint = sam_endpoint

def connect(reactor, control_endpoint=None, password_function=None):
    """
    Creates a :class:`txtorcon.Tor` instance by connecting to an
    already-running tor's control port. For example, a common default
    tor uses is UNIXClientEndpoint(reactor, '/var/run/tor/control') or
    TCP4ClientEndpoint(reactor, 'localhost', 9051)

    If only password authentication is available in the tor we connect
    to, the ``password_function`` is called (if supplied) to retrieve
    a valid password. This function can return a Deferred.

    For example::

        import txtorcon
        from twisted.internet.task import react
        from twisted.internet.defer import inlineCallbacks

        @inlineCallbacks
        def main(reactor):
            tor = yield txtorcon.connect(
                TCP4ClientEndpoint(reactor, "localhost", 9051)
            )
            state = yield tor.create_state()
            for circuit in state.circuits:
                print(circuit)

    :param control_endpoint: None, an IStreamClientEndpoint to connect
        to, or a Sequence of IStreamClientEndpoint instances to connect
        to. If None, a list of defaults are tried.

    :param password_function:
        See :class:`txtorcon.TorControlProtocol`

    :return:
        a Deferred that fires with a :class:`txtorcon.Tor` instance
    """

    @inlineCallbacks
    def try_endpoint(control_ep):
        assert IStreamClientEndpoint.providedBy(control_ep)
        proto = yield control_ep.connect(
            TorProtocolFactory(
                password_function=password_function
            )
        )
        config = yield TorConfig.from_protocol(proto)
        tor = Tor(reactor, proto, _tor_config=config)
        returnValue(tor)

    if control_endpoint is None:
        to_try = [
            UNIXClientEndpoint(reactor, '/var/run/tor/control'),
            TCP4ClientEndpoint(reactor, '127.0.0.1', 9051),
            TCP4ClientEndpoint(reactor, '127.0.0.1', 9151),
        ]
    elif IStreamClientEndpoint.providedBy(control_endpoint):
        to_try = [control_endpoint]
    elif isinstance(control_endpoint, Sequence):
        to_try = control_endpoint
        for ep in control_endpoint:
            if not IStreamClientEndpoint.providedBy(ep):
                raise ValueError(
                    "For control_endpoint=, '{}' must provide"
                    " IStreamClientEndpoint".format(ep)
                )
    else:
        raise ValueError(
            "For control_endpoint=, '{}' must provide"
            " IStreamClientEndpoint".format(control_endpoint)
        )

    errors = []
    for idx, ep in enumerate(to_try):
        try:
            tor = yield try_endpoint(ep)
            txtorlog.msg("Connected via '{}'".format(ep))
            returnValue(tor)
        except Exception as e:
            errors.append(e)
    if len(errors) == 1:
        raise errors[0]
    raise RuntimeError(
        'Failed to connect to: {}'.format(
            ', '.join(
                '{}: {}'.format(ep, err) for ep, err in zip(to_try, errors)
            )
        )
    )