class TorControlProtocol(LineOnlyReceiver):
    """
    This is the main class that talks to a Tor and implements the "raw"
    procotol.

    This instance does not track state; see :class:`txtorcon.TorState`
    for the current state of all Circuits, Streams and Routers.

    :meth:`txtorcon.TorState.build_circuit` allows you to build custom
    circuits.

    :meth:`txtorcon.TorControlProtocol.add_event_listener` can be used
    to listen for specific events.

    To see how circuit and stream listeners are used, see
    :class:`txtorcon.TorState`, which is also the place to go if you
    wish to add your own stream or circuit listeners.
    """

    def __init__(self, password_function=None):
        """
        :param password_function:
            A zero-argument callable which returns a password (or
            Deferred). It is only called if the Tor doesn't have
            COOKIE authentication turned on. Tor's default is COOKIE.
        """

        self.password_function = password_function
        """If set, a callable to query for a password to use for
        authentication to Tor (default is to use COOKIE, however). May
        return Deferred."""

        self._cookie_data = None
        """Data read from cookie file used to authenticate."""

        self.version = None
        """Version of Tor we've connected to."""

        self.is_owned = None
        """If not None, this is the PID of the Tor process we own
        (TAKEOWNERSHIP, etc)."""

        self.events = {}
        """events we've subscribed to (keyed by name like "GUARD", "STREAM")"""

        self.valid_events = {}
        """all valid events (name -> Event instance)"""

        self.valid_signals = []
        """A list of all valid signals we accept from Tor"""

        # NOTE: bad practice (and now deprecated) -- use
        # when_disconnected() instead
        self.on_disconnect = defer.Deferred()
        """
        This Deferred is triggered when the connection is closed. If there
        was an error, the errback is called instead. (Deprecated: use
        :func:`when_disconnected` instead)
        """

        self._when_disconnected = SingleObserver()
        """
        Internal use. A :class:`SingleObserver` for when_disconnected()
        """

        self._when_disconnected = SingleObserver()
        """
        Private. See :func:`.when_disconnected`
        """

        self.post_bootstrap = defer.Deferred()
        """
        This Deferred is triggered when we're done setting up
        (authentication, getting information from Tor). You will want
        to use this to do things with the :class:`TorControlProtocol`
        class when it's set up, like::

            def setup_complete(proto):
                print "Setup complete, attached to Tor version",proto.version

            def setup(proto):
                proto.post_bootstrap.addCallback(setup_complete)

            ep = TCP4ClientEndpoint(reactor, "localhost", 9051)
            ep.connect(TorProtocolFactory())
            d.addCallback(setup)

        See the helper method :func:`txtorcon.build_tor_connection`.
        """

        # variables related to the state machine
        self.defer = None        # Deferred we returned for the current command
        self.response = ''
        self.code = None
        self.command = None      # currently processing this command
        self.commands = []       # queued commands

        # Here we build up the state machine. Mostly it's pretty
        # simply, confounded by the fact that 600's (notify) can come
        # at any time AND can be multi-line itself. Luckily, these
        # can't be nested, nor can the responses be interleaved.

        idle = State("IDLE")
        recv = State("RECV")
        recvmulti = State("RECV_PLUS")
        recvnotify = State("NOTIFY_MULTILINE")

        idle.add_transition(Transition(idle,
                                       self._is_single_line_response,
                                       self._broadcast_response))
        idle.add_transition(Transition(recvmulti,
                                       self._is_multi_line,
                                       self._start_command))
        idle.add_transition(Transition(recv,
                                       self._is_continuation_line,
                                       self._start_command))

        recv.add_transition(Transition(recvmulti,
                                       self._is_multi_line,
                                       self._accumulate_response))
        recv.add_transition(Transition(recv,
                                       self._is_continuation_line,
                                       self._accumulate_response))
        recv.add_transition(Transition(idle,
                                       self._is_finish_line,
                                       self._broadcast_response))

        recvmulti.add_transition(Transition(recv,
                                            self._is_end_line,
                                            lambda x: None))
        recvmulti.add_transition(Transition(recvmulti,
                                            self._is_not_end_line,
                                            self._accumulate_multi_response))

        self.fsm = FSM([recvnotify, idle, recvmulti, recv])
        self.state_idle = idle
        # hand-set initial state default start state is first in the
        # list; the above looks nice in dotty though
        self.fsm.state = idle
        self.stop_debug()

    def start_debug(self):
        self.debuglog = open('txtorcon-debug.log', 'w')

    def stop_debug(self):
        def noop(*args, **kw):
            pass

        class NullLog(object):
            write = noop
            flush = noop
        self.debuglog = NullLog()

    def graphviz_data(self):
        return self.fsm.dotty()

    # see end of file for all the state machine matcher and
    # transition methods.

    def get_info_raw(self, *args):
        """
        Mostly for internal use; gives you the raw string back from the
        GETINFO command. See :meth:`getinfo
        <txtorcon.TorControlProtocol.get_info>`
        """
        return self.queue_command('GETINFO %s' % ' '.join(args))

    def get_info_incremental(self, key, line_cb):
        """
        Mostly for internal use; calls GETINFO for a single key and
        calls line_cb with each line received, as it is received.

        See :meth:`getinfo <txtorcon.TorControlProtocol.get_info>`
        """

        def strip_ok_and_call(line):
            if line.strip() != 'OK':
                line_cb(line)
        return self.queue_command('GETINFO %s' % key, strip_ok_and_call)

    # The following methods are the main TorController API and
    # probably the most interesting for users.

    def get_info(self, *args):
        """
        Uses GETINFO to obtain informatoin from Tor.

        :param args:
            should be a list or tuple of strings which are valid
            information keys. For valid keys, see control-spec.txt
            from torspec.

            .. todo:: make some way to automagically obtain valid
                keys, either from running Tor or parsing control-spec

        :return:
            a ``Deferred`` which will callback with a dict containing
            the keys you asked for. If you want to avoid the parsing
            into a dict, you can use get_info_raw instead.
        """
        d = self.get_info_raw(*args)
        d.addCallback(parse_keywords, key_hints=args)
        return d

    def get_info_single(self, key):
        """
        Uses GETINFO to obtain informatoin from Tor.

        :param key:
            the name of a GETINFO key to retrieve. For valid keys, see
            control-spec.txt from torspec.

        :return:
            a ``Deferred`` which will callback with the value for that
            key (a string).
        """
        d = self.get_info_raw(key)
        d.addCallback(parse_keywords, key_hints=[key])
        d.addCallback(lambda values: values[key])
        return d

    def get_conf(self, *args):
        """
        Uses GETCONF to obtain configuration values from Tor.

        :param args: any number of strings which are keys to get. To
            get all valid configuraiton names, you can call:
            ``get_info('config/names')``

        :return: a Deferred which callbacks with one or many
            configuration values (depends on what you asked for). See
            control-spec for valid keys (you can also use TorConfig which
            will come set up with all the keys that are valid). The value
            will be a dict.

        Note that Tor differentiates between an empty value and a
        default value; in the raw protocol one looks like '250
        MyFamily' versus '250 MyFamily=' where the latter is set to
        the empty string and the former is a default value. We
        differentiate these by setting the value in the dict to
        DEFAULT_VALUE for the default value case, or an empty string
        otherwise.
        """

        d = self.queue_command('GETCONF %s' % ' '.join(args))
        d.addCallback(parse_keywords).addErrback(log.err)
        return d

    def get_conf_single(self, key):
        """
        Uses GETCONF to obtain configuration values from Tor.

        :param key: a key whose CONF value to retrieve. To
            get all valid configuraiton names, you can call:
            ``get_info('config/names')``

        :return: a Deferred which callbacks with the configuration value.

        Note that Tor differentiates between an empty value and a
        default value; in the raw protocol one looks like '250
        MyFamily' versus '250 MyFamily=' where the latter is set to
        the empty string and the former is a default value. We
        differentiate these by returning DEFAULT_VALUE for the default
        value case, or an empty string otherwise.
        """

        d = self.queue_command('GETCONF {}'.format(key))
        d.addCallback(parse_keywords).addErrback(log.err)
        # d.addCallback(lambda kw: kw[key])  # extract key we asked for initially
        # ...but, the key can have a different string-name because Tor
        # will return *it's* representation (e.g. can ask for
        # SOCKSPORT but it will tell you SocksPort=9050)
        d.addCallback(lambda kw: list(kw.values())[0])
        return d

    def get_conf_raw(self, *args):
        """
        Same as get_conf, except that the results are not parsed into a dict
        """

        return self.queue_command('GETCONF %s' % ' '.join(args))

    def set_conf(self, *args):
        """
        set configuration values. see control-spec for valid
        keys. args is treated as a list containing name then value
        pairs. For example, ``set_conf('foo', 'bar')`` will (attempt
        to) set the key 'foo' to value 'bar'.

        :return: a ``Deferred`` that will callback with the response
            ('OK') or errback with the error code and message (e.g.
            ``"552 Unrecognized option: Unknown option 'foo'.  Failing."``)
        """
        if len(args) % 2:
            d = defer.Deferred()
            d.errback(RuntimeError("Expected an even number of arguments."))
            return d
        strargs = [str(x) for x in args]
        keys = [strargs[i] for i in range(0, len(strargs), 2)]
        values = [strargs[i] for i in range(1, len(strargs), 2)]

        def maybe_quote(s):
            if ' ' in s:
                return '"%s"' % s
            return s
        values = [maybe_quote(v) for v in values]
        args = ' '.join(map(lambda x, y: '%s=%s' % (x, y), keys, values))
        return self.queue_command('SETCONF ' + args)

    def signal(self, nm):
        """
        Issues a signal to Tor. See control-spec or
        :attr:`txtorcon.TorControlProtocol.valid_signals` for which ones
        are available and their return values.

        :return: a ``Deferred`` which callbacks with Tor's response
            (``OK`` or something like ``552 Unrecognized signal code "foo"``).
        """
        if nm not in self.valid_signals:
            raise RuntimeError("Invalid signal " + nm)
        return self.queue_command('SIGNAL %s' % nm)

    # XXX FIXME this should have been async all along :/
    def add_event_listener(self, evt, callback):
        """
        Add a listener to an Event object. This may be called multiple
        times for the same event. If it's the first listener, a new
        SETEVENTS call will be initiated to Tor.

        :param evt: event name, see also
            :attr:`txtorcon.TorControlProtocol.events` .keys(). These
            event names are queried from Tor (with `GETINFO events/names`)

        :param callback: any callable that takes a single argument
             which receives the text collected for the event from the
             tor control protocol.

        For more information on the events supported, see
        `control-spec section 4.1
        <https://gitweb.torproject.org/torspec.git/tree/control-spec.txt#n1260>`_

        .. note::
            this is a low-level interface; if you want to follow
            circuit or stream creation etc. see TorState and methods
            like add_circuit_listener

        :returns: a Deferred that fires when the listener is added
            (this may involve a controller command if this is the first
            listener for this event).

        .. todo::
            - should have an interface for the callback
            - show how to tie in Stem parsing if you want
        """

        if evt not in self.valid_events.values():
            try:
                evt = self.valid_events[evt]
            except KeyError:
                raise RuntimeError("Unknown event type: " + evt)

        if evt.name not in self.events:
            self.events[evt.name] = evt
            d = self.queue_command('SETEVENTS %s' % ' '.join(self.events.keys()))
        else:
            d = defer.succeed(None)
        evt.listen(callback)
        return d

    # XXX this should have been async all along
    def remove_event_listener(self, evt, cb):
        """
        The opposite of :meth:`TorControlProtocol.add_event_listener`

        :param evt: the event name (or an Event object)

        :param cb: the callback object to remove

        :returns: a Deferred that fires when the listener is removed
            (this may involve a controller command if this is the last
            listener for this event).
        """
        if evt not in self.valid_events.values():
            # this lets us pass a string or a real event-object
            try:
                evt = self.valid_events[evt]
            except KeyError:
                raise RuntimeError("Unknown event type: " + evt)

        evt.unlisten(cb)
        if len(evt.callbacks) == 0:
            # note there's a slight window here for an event of this
            # type to come in before the SETEVENTS succeeds; see
            # _handle_notify which explicitly ignore this case.
            del self.events[evt.name]
            return self.queue_command('SETEVENTS %s' % ' '.join(self.events.keys()))
        else:
            return defer.succeed(None)

    def protocolinfo(self):
        """
        :return: a Deferred which will give you PROTOCOLINFO; see control-spec
        """

        return self.queue_command("PROTOCOLINFO 1")

    def authenticate(self, passphrase):
        """
        Call the AUTHENTICATE command.

        Quoting torspec/control-spec.txt: "The authentication token
        can be specified as either a quoted ASCII string, or as an
        unquoted hexadecimal encoding of that same string (to avoid
        escaping issues)."
        """
        if not isinstance(passphrase, bytes):
            passphrase = passphrase.encode()
        phrase = b2a_hex(passphrase)
        return self.queue_command(b'AUTHENTICATE ' + phrase)

    def quit(self):
        """
        Sends the QUIT command, which asks Tor to hang up on this
        controller connection.

        If you've taken ownership of the Tor to which you're
        connected, this should also cause it to exit. Otherwise, it
        won't.
        """
        return self.queue_command('QUIT')

    def queue_command(self, cmd, arg=None):
        """
        returns a Deferred which will fire with the response data when
        we get it

        Note that basically every request is ultimately funelled
        through this command.
        """

        if not isinstance(cmd, bytes):
            cmd = cmd.encode('ascii')
        d = defer.Deferred()
        self.commands.append((d, cmd, arg))
        self._maybe_issue_command()
        return d

    def when_disconnected(self):
        """
        :returns: a Deferred that fires when (if) we disconnect from our
            Tor process.
        """
        return self._when_disconnected.when_fired()

    # the remaining methods are internal API implementations,
    # callbacks and state-tracking methods -- you shouldn't have any
    # need to call them.

    def lineReceived(self, line):
        """
        :api:`twisted.protocols.basic.LineOnlyReceiver` API
        """

        self.debuglog.write(line + b'\n')
        self.debuglog.flush()
        self.fsm.process(line.decode('ascii'))

    def connectionMade(self):
        "Protocol API"
        txtorlog.msg('got connection, authenticating')
        # XXX this Deferred is just being dropped on the floor
        d = self.protocolinfo()
        d.addCallback(self._do_authenticate)
        d.addErrback(self._auth_failed)

    def connectionLost(self, reason):
        "Protocol API"
        txtorlog.msg('connection terminated: ' + str(reason))
        if reason.check(ConnectionDone):
            self._when_disconnected.fire(self)
        else:
            self._when_disconnected.fire(reason)

        # ...and this is why we don't do on_disconnect = Deferred() :(
        # and instead should have had on_disconnect() method that
        # returned a new Deferred to each caller..(we're checking if
        # this Deferred has any callbacks because if it doesn't we'll
        # generate an "Unhandled error in Deferred")
        if self.on_disconnect and not self.on_disconnect.called and self.on_disconnect.callbacks:
            warn(
                'TorControlProtocol.on_disconnect is deprecated; use .when_disconnected() instead',
                DeprecationWarning,
            )
            if reason.check(ConnectionDone):
                self.on_disconnect.callback(self)
            else:
                self.on_disconnect.errback(reason)
        self.on_disconnect = None
        self._when_disconnected.fire(self)
        outstanding = [self.command] + self.commands if self.command else self.commands
        for d, cmd, cmd_arg in outstanding:
            if not d.called:
                d.errback(
                    Failure(
                        TorDisconnectError(
                            text=("Tor unexpectedly disconnected while "
                                  "running: {}".format(cmd.decode('ascii'))),
                            error=reason,
                        )
                    )
                )
        return None

    def _handle_notify(self, code, rest):
        """
        Internal method to deal with 600-level responses.
        """

        firstline = rest[:rest.find('\n')]
        args = firstline.split()
        name = args[0]
        if name in self.events:
            self.events[name].got_update(rest[len(name) + 1:])
            return
        # not considering this an error, as there's a slight window
        # after remove_event_listener is called (so the handler is
        # deleted) but the SETEVENTS command has not yet succeeded

    def _maybe_issue_command(self):
        """
        If there's at least one command queued and we're not currently
        processing a command, this will issue the next one on the
        wire.
        """
        if self.command:
            return

        if len(self.commands):
            self.command = self.commands.pop(0)
            (d, cmd, cmd_arg) = self.command
            self.defer = d

            self.debuglog.write(cmd + b'\n')
            self.debuglog.flush()

            data = cmd + b'\r\n'
            txtorlog.msg("cmd: {}".format(data.strip()))
            self.transport.write(data)

    def _auth_failed(self, fail):
        """
        Errback if authentication fails.
        """
        # it should be impossible for post_bootstrap to be
        # already-called/failed at this point; _auth_failed only stems
        # from _do_authentication failing
        self.post_bootstrap.errback(fail)
        return None

    def _safecookie_authchallenge(self, reply):
        """
        Callback on AUTHCHALLENGE SAFECOOKIE
        """
        if self._cookie_data is None:
            raise RuntimeError("Cookie data not read.")
        kw = parse_keywords(reply.replace(' ', '\n'))

        server_hash = base64.b16decode(kw['SERVERHASH'])
        server_nonce = base64.b16decode(kw['SERVERNONCE'])
        # FIXME put string in global. or something.
        expected_server_hash = hmac_sha256(
            b"Tor safe cookie authentication server-to-controller hash",
            self._cookie_data + self.client_nonce + server_nonce,
        )

        if not compare_via_hash(expected_server_hash, server_hash):
            raise RuntimeError(
                'Server hash not expected; wanted "%s" and got "%s".' %
                (base64.b16encode(expected_server_hash),
                 base64.b16encode(server_hash))
            )

        client_hash = hmac_sha256(
            b"Tor safe cookie authentication controller-to-server hash",
            self._cookie_data + self.client_nonce + server_nonce
        )
        client_hash_hex = base64.b16encode(client_hash)
        return self.queue_command(b'AUTHENTICATE ' + client_hash_hex)

    def _read_cookie(self, cookiefile):
        """
        Open and read a cookie file
        :param cookie: Path to the cookie file
        """
        self._cookie_data = None
        self._cookie_data = open(cookiefile, 'rb').read()
        if len(self._cookie_data) != 32:
            raise RuntimeError(
                "Expected authentication cookie to be 32 bytes, got %d" %
                len(self._cookie_data)
            )

    def _do_authenticate(self, protoinfo):
        """
        Callback on PROTOCOLINFO to actually authenticate once we know
        what's supported.
        """
        methods = None
        cookie_auth = False
        for line in protoinfo.split('\n'):
            if line[:5] == 'AUTH ':
                kw = parse_keywords(line[5:].replace(' ', '\n'))
                methods = kw['METHODS'].split(',')
        if not methods:
            raise RuntimeError(
                "Didn't find AUTH line in PROTOCOLINFO response."
            )

        if 'SAFECOOKIE' in methods or 'COOKIE' in methods:
            cookiefile_match = re.search(r'COOKIEFILE=("(?:[^"\\]|\\.)*")',
                                         protoinfo)
            if cookiefile_match:
                cookiefile = cookiefile_match.group(1)
                cookiefile = unescape_quoted_string(cookiefile)
                try:
                    self._read_cookie(cookiefile)
                    cookie_auth = True
                except IOError as why:
                    txtorlog.msg("Reading COOKIEFILE failed: " + str(why))
                    if self.password_function and 'HASHEDPASSWORD' in methods:
                        txtorlog.msg("Falling back to password")
                    else:
                        raise RuntimeError(
                            "Failed to read COOKIEFILE '{fname}': {msg}\n".format(
                                fname=cookiefile,
                                msg=str(why),
                            )
                            # "On Debian, join the debian-tor group"
                        )
            else:
                txtorlog.msg("Didn't get COOKIEFILE")
                raise RuntimeError(
                    "Got 'COOKIE' or 'SAFECOOKIE' method, but no 'COOKIEFILE'"
                )

        if cookie_auth:
            if 'SAFECOOKIE' in methods:
                txtorlog.msg("Using SAFECOOKIE authentication", cookiefile,
                             len(self._cookie_data), "bytes")
                self.client_nonce = os.urandom(32)

                cmd = b'AUTHCHALLENGE SAFECOOKIE ' + \
                      hexlify(self.client_nonce)
                d = self.queue_command(cmd)
                d.addCallback(self._safecookie_authchallenge)
                d.addCallback(self._bootstrap)
                return d

            elif 'COOKIE' in methods:
                txtorlog.msg("Using COOKIE authentication",
                             cookiefile, len(self._cookie_data), "bytes")
                d = self.authenticate(self._cookie_data)
                d.addCallback(self._bootstrap)
                return d

        if self.password_function and 'HASHEDPASSWORD' in methods:
            d = defer.maybeDeferred(self.password_function)
            d.addCallback(maybe_coroutine)
            d.addCallback(self._do_password_authentication)
            return d

        if 'NULL' in methods:
            d = self.queue_command('AUTHENTICATE')
            d.addCallback(self._bootstrap)
            return d

        return defer.fail(
            RuntimeError(
                "The Tor I connected to doesn't support SAFECOOKIE nor COOKIE"
                " authentication (or we can't read the cookie files) and I have"
                " no password_function specified."
            )
        )

    def _do_password_authentication(self, passwd):
        if not passwd:
            raise RuntimeError("No password available.")
        d = self.authenticate(passwd)
        d.addCallback(self._bootstrap)
        d.addErrback(self._auth_failed)

    def _set_valid_events(self, events):
        "used as a callback; see _bootstrap"
        self.valid_events = {}
        for x in events.split():
            self.valid_events[x] = Event(x)

    @defer.inlineCallbacks
    def _bootstrap(self, *args):
        """
        The inlineCallbacks decorator allows us to make this method
        look synchronous; see the Twisted docs. Each yeild is for a
        Deferred after which the method continues. When this method
        finally exits, we're set up and do the post_bootstrap
        callback.
        """

        try:
            self.valid_signals = yield self.get_info('signal/names')
            self.valid_signals = self.valid_signals['signal/names']
        except TorProtocolError:
            self.valid_signals = ["RELOAD", "DUMP", "DEBUG", "NEWNYM",
                                  "CLEARDNSCACHE"]

        self.version = yield self.get_info('version')
        self.version = self.version['version']
        txtorlog.msg("Connected to a Tor with VERSION", self.version)
        eventnames = yield self.get_info('events/names')
        eventnames = eventnames['events/names']
        self._set_valid_events(eventnames)

        yield self.queue_command('USEFEATURE EXTENDED_EVENTS')

        self.post_bootstrap.callback(self)
        defer.returnValue(self)

    # State Machine transitions and matchers. See the __init__ method
    # for a way to output a GraphViz dot diagram of the machine.

    def _is_end_line(self, line):
        "for FSM"
        return line.strip() == '.'

    def _is_not_end_line(self, line):
        "for FSM"
        return not self._is_end_line(line)

    def _is_single_line_response(self, line):
        "for FSM"
        try:
            code = int(line[:3])
        except Exception:
            return False

        sl = len(line) > 3 and line[3] == ' '
        # print "single line?",line,sl
        if sl:
            self.code = code
            return True
        return False

    def _start_command(self, line):
        "for FSM"
        # print "startCommand",self.code,line
        self.code = int(line[:3])
        # print "startCommand:",self.code
        if self.command and self.command[2] is not None:
            self.command[2](line[4:])
        else:
            self.response = line[4:] + '\n'
        return None

    def _is_continuation_line(self, line):
        "for FSM"
        code = int(line[:3])
        if self.code and self.code != code:
            raise RuntimeError("Unexpected code %d, wanted %d" % (code,
                                                                  self.code))
        return line[3] == '-'

    def _is_multi_line(self, line):
        "for FSM"
        code = int(line[:3])
        if self.code and self.code != code:
            raise RuntimeError("Unexpected code %d, wanted %d" % (code,
                                                                  self.code))
        return line[3] == '+'

    def _accumulate_multi_response(self, line):
        "for FSM"
        if self.command and self.command[2] is not None:
            self.command[2](line)

        else:
            self.response += (line + '\n')
        return None

    def _accumulate_response(self, line):
        "for FSM"
        if self.command and self.command[2] is not None:
            self.command[2](line[4:])

        else:
            self.response += (line[4:] + '\n')
        return None

    def _is_finish_line(self, line):
        "for FSM"
        # print "isFinish",line
        if len(line) < 1:
            return False
        if line[0] == '.':
            return True
        if len(line) > 3 and line[3] == ' ':
            return True
        return False

    def _broadcast_response(self, line):
        "for FSM"
        if len(line) > 3:
            if self.code >= 200 and self.code < 300 and \
               self.command and self.command[2] is not None:
                self.command[2](line[4:])
                resp = ''

            else:
                resp = self.response + line[4:]
        else:
            resp = self.response
        self.response = ''
        if self.code is None:
            raise RuntimeError("No code set yet in broadcast response.")
        elif self.code >= 200 and self.code < 300:
            if self.defer is None:
                raise RuntimeError(
                    'Got a response, but didn\'t issue a command: "%s"' % resp
                )
            if resp.endswith('\nOK'):
                resp = resp[:-3]
            self.defer.callback(resp)
        elif self.code >= 500 and self.code < 600:
            err = TorProtocolError(self.code, resp)
            self.defer.errback(err)
        elif self.code >= 600 and self.code < 700:
            self._handle_notify(self.code, resp)
            self.code = None
            return
        else:
            raise RuntimeError(
                "Unknown code in broadcast response %d." % self.code
            )

        # note: we don't do this for 600-level responses
        self.command = None
        self.code = None
        self.defer = None
        self._maybe_issue_command()
        return None class TorConfig(object):
    """This class abstracts out Tor's config, and can be used both to
    create torrc files from nothing and track live configuration of a Tor
    instance.

    Also, it gives easy access to all the configuration options
    present. This is initialized at "bootstrap" time, providing
    attribute-based access thereafter. Note that after you set some
    number of items, you need to do a save() before these are sent to
    Tor (and then they will be done as one SETCONF).

    You may also use this class to construct a configuration from
    scratch (e.g. to give to :func:`txtorcon.launch_tor`). In this
    case, values are reflected right away. (If we're not bootstrapped
    to a Tor, this is the mode).

    Note that you do not need to call save() if you're just using
    TorConfig to create a .torrc file or for input to launch_tor().

    This class also listens for CONF_CHANGED events to update the
    cached data in the event other controllers (etc) changed it.

    There is a lot of magic attribute stuff going on in here (which
    might be a bad idea, overall) but the *intent* is that you can
    just set Tor options and it will all Just Work. For config items
    that take multiple values, set that to a list. For example::

        conf = TorConfig(...)
        conf.SOCKSPort = [9050, 1337]
        conf.HiddenServices.append(HiddenService(...))

    (Incoming objects, like lists, are intercepted and wrapped).

    FIXME: when is CONF_CHANGED introduced in Tor? Can we do anything
    like it for prior versions?

    FIXME:

        - HiddenServiceOptions is special: GETCONF on it returns
          several (well, two) values. Besides adding the two keys 'by
          hand' do we need to do anything special? Can't we just depend
          on users doing 'conf.hiddenservicedir = foo' AND
          'conf.hiddenserviceport = bar' before a save() ?

        - once I determine a value is default, is there any way to
          actually get what this value is?

    """

    @staticmethod
    @defer.inlineCallbacks
    def from_protocol(proto):
        """
        This creates and returns a ready-to-go TorConfig instance from the
        given protocol, which should be an instance of
        TorControlProtocol.
        """
        cfg = TorConfig(control=proto)
        yield cfg.post_bootstrap
        defer.returnValue(cfg)

    def __init__(self, control=None):
        self.config = {}
        '''Current configuration, by keys.'''

        if control is None:
            self._protocol = None
            self.__dict__['_accept_all_'] = None

        else:
            self._protocol = ITorControlProtocol(control)

        self.unsaved = OrderedDict()
        '''Configuration that has been changed since last save().'''

        self.parsers = {}
        '''Instances of the parser classes, subclasses of TorConfigType'''

        self.list_parsers = set(['hiddenservices', 'ephemeralonionservices'])
        '''All the names (keys from .parsers) that are a List of something.'''

        # during bootstrapping we decide whether we support the
        # following features. A thing goes in here if TorConfig
        # behaves differently depending upon whether it shows up in
        # "GETINFO config/names"
        self._supports = dict(
            HiddenServiceDirGroupReadable=False
        )
        self._defaults = dict()

        self.post_bootstrap = defer.Deferred()
        if self.protocol:
            if self.protocol.post_bootstrap:
                self.protocol.post_bootstrap.addCallback(
                    self.bootstrap).addErrback(self.post_bootstrap.errback)
            else:
                self.bootstrap()

        else:
            self.do_post_bootstrap(self)

        self.__dict__['_setup_'] = None

    def socks_endpoint(self, reactor, port=None):
        """
        Returns a TorSocksEndpoint configured to use an already-configured
        SOCKSPort from the Tor we're connected to. By default, this
        will be the very first SOCKSPort.

        :param port: a str, the first part of the SOCKSPort line (that
            is, a port like "9151" or a Unix socket config like
            "unix:/path". You may also specify a port as an int.

        If you need to use a particular port that may or may not
        already be configured, see the async method
        :meth:`txtorcon.TorConfig.create_socks_endpoint`
        """

        if len(self.SocksPort) == 0:
            raise RuntimeError(
                "No SOCKS ports configured"
            )

        socks_config = None
        if port is None:
            socks_config = self.SocksPort[0]
        else:
            port = str(port)  # in case e.g. an int passed in
            if ' ' in port:
                raise ValueError(
                    "Can't specify options; use create_socks_endpoint instead"
                )

            for idx, port_config in enumerate(self.SocksPort):
                # "SOCKSPort" is a gnarly beast that can have a bunch
                # of options appended, so we have to split off the
                # first thing which *should* be the port (or can be a
                # string like 'unix:')
                if port_config.split()[0] == port:
                    socks_config = port_config
                    break
        if socks_config is None:
            raise RuntimeError(
                "No SOCKSPort configured for port {}".format(port)
            )

        return _endpoint_from_socksport_line(reactor, socks_config)

    @defer.inlineCallbacks
    def create_socks_endpoint(self, reactor, socks_config):
        """
        Creates a new TorSocksEndpoint instance given a valid
        configuration line for ``SocksPort``; if this configuration
        isn't already in the underlying tor, we add it. Note that this
        method may call :meth:`txtorcon.TorConfig.save()` on this instance.

        Note that calling this with `socks_config=None` is equivalent
        to calling `.socks_endpoint` (which is not async).

        XXX socks_config should be .. i dunno, but there's fucking
        options and craziness, e.g. default Tor Browser Bundle is:
        ['9150 IPv6Traffic PreferIPv6 KeepAliveIsolateSOCKSAuth',
        '9155']

        XXX maybe we should say "socks_port" as the 3rd arg, insist
        it's an int, and then allow/support all the other options
        (e.g. via kwargs)

        XXX we could avoid the "maybe call .save()" thing; worth it?
        (actually, no we can't or the Tor won't have it config'd)
        """

        yield self.post_bootstrap

        if socks_config is None:
            if len(self.SocksPort) == 0:
                raise RuntimeError(
                    "socks_port is None and Tor has no SocksPorts configured"
                )
            socks_config = self.SocksPort[0]
        else:
            if not any([socks_config in port for port in self.SocksPort]):
                # need to configure Tor
                self.SocksPort.append(socks_config)
                try:
                    yield self.save()
                except TorProtocolError as e:
                    extra = ''
                    if socks_config.startswith('unix:'):
                        # XXX so why don't we check this for the
                        # caller, earlier on?
                        extra = '\nNote Tor has specific ownership/permissions ' +\
                                'requirements for unix sockets and parent dir.'
                    raise RuntimeError(
                        "While configuring SOCKSPort to '{}', error from"
                        " Tor: {}{}".format(
                            socks_config, e, extra
                        )
                    )

        defer.returnValue(
            _endpoint_from_socksport_line(reactor, socks_config)
        )

    # FIXME should re-name this to "tor_protocol" to be consistent
    # with other things? Or rename the other things?
    """
    read-only access to TorControlProtocol. Call attach_protocol() to
    set it, which can only be done if we don't already have a
    protocol.
    """
    def _get_protocol(self):
        return self.__dict__['_protocol']
    protocol = property(_get_protocol)
    tor_protocol = property(_get_protocol)

    def attach_protocol(self, proto):
        """
        returns a Deferred that fires once we've set this object up to
        track the protocol. Fails if we already have a protocol.
        """
        if self._protocol is not None:
            raise RuntimeError("Already have a protocol.")
        # make sure we have nothing in self.unsaved
        self.save()
        self.__dict__['_protocol'] = proto

        # FIXME some of this is duplicated from ctor
        del self.__dict__['_accept_all_']
        self.__dict__['post_bootstrap'] = defer.Deferred()
        if proto.post_bootstrap:
            proto.post_bootstrap.addCallback(self.bootstrap)
        return self.__dict__['post_bootstrap']

    def __setattr__(self, name, value):
        """
        we override this so that we can provide direct attribute
        access to our config items, and move them into self.unsaved
        when they've been changed. hiddenservices have to be special
        unfortunately. the _setup_ thing is so that we can set up the
        attributes we need in the constructor without uusing __dict__
        all over the place.
        """

        # appease flake8's hatred of lambda :/
        def has_setup_attr(o):
            return '_setup_' in o.__dict__

        def has_accept_all_attr(o):
            return '_accept_all_' in o.__dict__

        def is_hidden_services(s):
            return s.lower() == "hiddenservices"

        if has_setup_attr(self):
            name = self._find_real_name(name)
            if not has_accept_all_attr(self) and not is_hidden_services(name):
                value = self.parsers[name].validate(value, self, name)
            if isinstance(value, list):
                value = _ListWrapper(
                    value, functools.partial(self.mark_unsaved, name))

            name = self._find_real_name(name)
            self.unsaved[name] = value

        else:
            super(TorConfig, self).__setattr__(name, value)

    def _maybe_create_listwrapper(self, rn):
        if rn.lower() in self.list_parsers and rn not in self.config:
            self.config[rn] = _ListWrapper([], functools.partial(
                self.mark_unsaved, rn))

    def __getattr__(self, name):
        """
        on purpose, we don't return self.unsaved if the key is in there
        because I want the config to represent the running Tor not
        ``things which might get into the running Tor if save() were
        to be called''
        """
        rn = self._find_real_name(name)
        if '_accept_all_' in self.__dict__ and rn in self.unsaved:
            return self.unsaved[rn]
        self._maybe_create_listwrapper(rn)
        v = self.config[rn]
        if v == DEFAULT_VALUE:
            v = self.__dict__['_defaults'].get(rn, DEFAULT_VALUE)
        return v

    def __contains__(self, item):
        if item in self.unsaved and '_accept_all_' in self.__dict__:
            return True
        return item in self.config

    def __iter__(self):
        '''
        FIXME needs proper iterator tests in test_torconfig too
        '''
        for x in self.config.__iter__():
            yield x
        for x in self.__dict__['unsaved'].__iter__():
            yield x

    def get_type(self, name):
        """
        return the type of a config key.

        :param: name the key

        FIXME can we do something more-clever than this for client
        code to determine what sort of thing a key is?
        """

        # XXX FIXME uhm...how to do all the different types of hidden-services?
        if name.lower() == 'hiddenservices':
            return FilesystemOnionService
        return type(self.parsers[name])

    def _conf_changed(self, arg):
        """
        internal callback. from control-spec:

        4.1.18. Configuration changed

          The syntax is:
             StartReplyLine *(MidReplyLine) EndReplyLine

             StartReplyLine = "650-CONF_CHANGED" CRLF
             MidReplyLine = "650-" KEYWORD ["=" VALUE] CRLF
             EndReplyLine = "650 OK"

          Tor configuration options have changed (such as via a SETCONF or
          RELOAD signal). KEYWORD and VALUE specify the configuration option
          that was changed.  Undefined configuration options contain only the
          KEYWORD.
        """

        conf = parse_keywords(arg, multiline_values=False)
        for (k, v) in conf.items():
            # v will be txtorcon.DEFAULT_VALUE already from
            # parse_keywords if it was unspecified
            real_name = self._find_real_name(k)
            if real_name in self.parsers:
                v = self.parsers[real_name].parse(v)
            self.config[real_name] = v

    def bootstrap(self, arg=None):
        '''
        This only takes args so it can be used as a callback. Don't
        pass an arg, it is ignored.
        '''
        try:
            d = self.protocol.add_event_listener(
                'CONF_CHANGED', self._conf_changed)
        except RuntimeError:
            # for Tor versions which don't understand CONF_CHANGED
            # there's nothing we can really do.
            log.msg(
                "Can't listen for CONF_CHANGED event; won't stay up-to-date "
                "with other clients.")
            d = defer.succeed(None)
        d.addCallback(lambda _: self.protocol.get_info_raw("config/names"))
        d.addCallback(self._do_setup)
        d.addCallback(self.do_post_bootstrap)
        d.addErrback(self.do_post_errback)

    def do_post_errback(self, f):
        self.post_bootstrap.errback(f)
        return None

    def do_post_bootstrap(self, arg):
        if not self.post_bootstrap.called:
            self.post_bootstrap.callback(self)
        return self

    def needs_save(self):
        return len(self.unsaved) > 0

    def mark_unsaved(self, name):
        name = self._find_real_name(name)
        if name in self.config and name not in self.unsaved:
            self.unsaved[name] = self.config[self._find_real_name(name)]

    def save(self):
        """
        Save any outstanding items. This returns a Deferred which will
        errback if Tor was unhappy with anything, or callback with
        this TorConfig object on success.
        """

        if not self.needs_save():
            return defer.succeed(self)

        args = []
        directories = []
        for (key, value) in self.unsaved.items():
            if key == 'HiddenServices':
                self.config['HiddenServices'] = value
                # using a list here because at least one unit-test
                # cares about order -- and conceivably order *could*
                # matter here, to Tor...
                services = list()
                # authenticated services get flattened into the HiddenServices list...
                for hs in value:
                    if IOnionClient.providedBy(hs):
                        parent = IOnionClient(hs).parent
                        if parent not in services:
                            services.append(parent)
                    elif isinstance(hs, (EphemeralOnionService, EphemeralHiddenService)):
                        raise ValueError(
                            "Only filesystem based Onion services may be added"
                            " via TorConfig.hiddenservices; ephemeral services"
                            " must be created with 'create_onion_service'."
                        )
                    else:
                        if hs not in services:
                            services.append(hs)

                for hs in services:
                    for (k, v) in hs.config_attributes():
                        if k == 'HiddenServiceDir':
                            if v not in directories:
                                directories.append(v)
                                args.append(k)
                                args.append(v)
                            else:
                                raise RuntimeError("Trying to add hidden service with same HiddenServiceDir: %s" % v)
                        else:
                            args.append(k)
                            args.append(v)
                continue

            if isinstance(value, list):
                for x in value:
                    # FIXME XXX
                    if x is not DEFAULT_VALUE:
                        args.append(key)
                        args.append(str(x))

            else:
                args.append(key)
                args.append(value)

            # FIXME in future we should wait for CONF_CHANGED and
            # update then, right?
            real_name = self._find_real_name(key)
            if not isinstance(value, list) and real_name in self.parsers:
                value = self.parsers[real_name].parse(value)
            self.config[real_name] = value

        # FIXME might want to re-think this, but currently there's no
        # way to put things into a config and get them out again
        # nicely...unless you just don't assign a protocol
        if self.protocol:
            d = self.protocol.set_conf(*args)
            d.addCallback(self._save_completed)
            return d

        else:
            self._save_completed()
            return defer.succeed(self)

    def _save_completed(self, *args):
        '''internal callback'''
        self.__dict__['unsaved'] = {}
        return self

    def _find_real_name(self, name):
        keys = list(self.__dict__['parsers'].keys()) + list(self.__dict__['config'].keys())
        for x in keys:
            if x.lower() == name.lower():
                return x
        return name

    @defer.inlineCallbacks
    def _get_defaults(self):
        try:
            defaults_raw = yield self.protocol.get_info_raw("config/defaults")
            defaults = {}
            for line in defaults_raw.split('\n')[1:]:
                k, v = line.split(' ', 1)
                if k in defaults:
                    if isinstance(defaults[k], list):
                        defaults[k].append(v)
                    else:
                        defaults[k] = [defaults[k], v]
                else:
                    defaults[k] = v
        except TorProtocolError:
            # must be a version of Tor without config/defaults
            defaults = dict()
        defer.returnValue(defaults)

    @defer.inlineCallbacks
    def _do_setup(self, data):
        defaults = self.__dict__['_defaults'] = yield self._get_defaults()

        for line in data.split('\n'):
            if line == "config/names=":
                continue

            (name, value) = line.split()
            if name in self._supports:
                self._supports[name] = True

            if name == 'HiddenServiceOptions':
                # set up the "special-case" hidden service stuff
                servicelines = yield self.protocol.get_conf_raw(
                    'HiddenServiceOptions')
                self._setup_hidden_services(servicelines)
                continue

            # there's a whole bunch of FooPortLines (where "Foo" is
            # "Socks", "Control", etc) and some have defaults, some
            # don't but they all have FooPortLines, FooPort, and
            # __FooPort definitions so we only "do stuff" for the
            # "FooPortLines"
            if name.endswith('PortLines'):
                rn = self._find_real_name(name[:-5])
                self.parsers[rn] = String()  # not Port() because options etc
                self.list_parsers.add(rn)
                v = yield self.protocol.get_conf(name[:-5])
                v = v[name[:-5]]

                initial = []
                if v == DEFAULT_VALUE or v == 'auto':
                    try:
                        initial = defaults[name[:-5]]
                    except KeyError:
                        default_key = '__{}'.format(name[:-5])
                        default = yield self.protocol.get_conf_single(default_key)
                        if not default:
                            initial = []
                        else:
                            initial = [default]
                else:
                    initial = [self.parsers[rn].parse(v)]
                self.config[rn] = _ListWrapper(
                    initial, functools.partial(self.mark_unsaved, rn))

            # XXX for Virtual check that it's one of the *Ports things
            # (because if not it should be an error)
            if value in ('Dependant', 'Dependent', 'Virtual'):
                continue

            # there's a thing called "Boolean+Auto" which is -1 for
            # auto, 0 for false and 1 for true. could be nicer if it
            # was called AutoBoolean or something, but...
            value = value.replace('+', '_')

            inst = None
            # FIXME: put parser classes in dict instead?
            for cls in config_types:
                if cls.__name__ == value:
                    inst = cls()
            if not inst:
                raise RuntimeError("Don't have a parser for: " + value)
            v = yield self.protocol.get_conf(name)
            v = v[name]

            rn = self._find_real_name(name)
            self.parsers[rn] = inst
            if is_list_config_type(inst.__class__):
                self.list_parsers.add(rn)
                parsed = self.parsers[rn].parse(v)
                if parsed == [DEFAULT_VALUE]:
                    parsed = defaults.get(rn, [])
                self.config[rn] = _ListWrapper(
                    parsed, functools.partial(self.mark_unsaved, rn))

            else:
                if v == '' or v == DEFAULT_VALUE:
                    parsed = self.parsers[rn].parse(defaults.get(rn, DEFAULT_VALUE))
                else:
                    parsed = self.parsers[rn].parse(v)
                self.config[rn] = parsed

        # get any ephemeral services we own, or detached services.
        # these are *not* _ListWrappers because we don't care if they
        # change, nothing in Tor's config exists for these (probably
        # begging the question: why are we putting them in here at all
        # then...?)
        try:
            ephemeral = yield self.protocol.get_info('onions/current')
        except Exception:
            self.config['EphemeralOnionServices'] = []
        else:
            onions = []
            for line in ephemeral['onions/current'].split('\n'):
                onion = line.strip()
                if onion:
                    onions.append(
                        EphemeralOnionService(
                            self,
                            ports=[],  # no way to discover ports=
                            hostname=onion,
                            private_key=DISCARD,  # we don't know it, anyway
                            version=2,
                            detach=False,
                        )
                    )
            self.config['EphemeralOnionServices'] = onions

        try:
            detached = yield self.protocol.get_info('onions/detached')
        except Exception:
            self.config['DetachedOnionServices'] = []
        else:
            onions = []
            for line in detached['onions/detached'].split('\n'):
                onion = line.strip()
                if onion:
                    onions.append(
                        EphemeralOnionService(
                            self,
                            ports=[],  # no way to discover original ports=
                            hostname=onion,
                            detach=True,
                            private_key=DISCARD,
                        )
                    )
            self.config['DetachedOnionServices'] = onions
        defer.returnValue(self)

    def _setup_hidden_services(self, servicelines):

        def maybe_add_hidden_service():
            if directory is not None:
                if directory not in directories:
                    directories.append(directory)
                    if not auth:
                        service = FilesystemOnionService(
                            self, directory, ports, ver, group_read
                        )
                        hs.append(service)
                    else:
                        auth_type, clients = auth.split(' ', 1)
                        clients = clients.split(',')
                        if auth_type == 'basic':
                            auth0 = AuthBasic(clients)
                        elif auth_type == 'stealth':
                            auth0 = AuthStealth(clients)
                        else:
                            raise ValueError(
                                "Unknown auth type '{}'".format(auth_type)
                            )
                        parent_service = FilesystemAuthenticatedOnionService(
                            self, directory, ports, auth0, ver, group_read
                        )
                        for client_name in parent_service.client_names():
                            hs.append(parent_service.get_client(client_name))
                else:
                    raise RuntimeError("Trying to add hidden service with same HiddenServiceDir: %s" % directory)

        hs = []
        directory = None
        directories = []
        ports = []
        ver = None
        group_read = None
        auth = None
        for line in servicelines.split('\n'):
            if not len(line.strip()):
                continue

            if line == 'HiddenServiceOptions':
                continue
            k, v = line.split('=')
            if k == 'HiddenServiceDir':
                maybe_add_hidden_service()
                directory = v
                _directory = directory
                directory = os.path.abspath(directory)
                if directory != _directory:
                    warnings.warn(
                        "Directory path: %s changed to absolute path: %s" % (_directory, directory),
                        RuntimeWarning
                    )
                ports = []
                ver = None
                auth = None
                group_read = 0

            elif k == 'HiddenServicePort':
                ports.append(v)

            elif k == 'HiddenServiceVersion':
                ver = int(v)

            elif k == 'HiddenServiceAuthorizeClient':
                if auth is not None:
                    # definitely error, or keep going?
                    raise ValueError("Multiple HiddenServiceAuthorizeClient lines for one service")
                auth = v

            elif k == 'HiddenServiceDirGroupReadable':
                group_read = int(v)

            else:
                raise RuntimeError("Can't parse HiddenServiceOptions: " + k)

        maybe_add_hidden_service()

        name = 'HiddenServices'
        self.config[name] = _ListWrapper(
            hs, functools.partial(self.mark_unsaved, name))

    def config_args(self):
        '''
        Returns an iterator of 2-tuples (config_name, value), one for each
        configuration option in this config. This is more-or-less an
        internal method, but see, e.g., launch_tor()'s implementation
        if you think you need to use this for something.

        See :meth:`txtorcon.TorConfig.create_torrc` which returns a
        string which is also a valid ``torrc`` file
        '''

        everything = dict()
        everything.update(self.config)
        everything.update(self.unsaved)

        for (k, v) in list(everything.items()):
            if type(v) is _ListWrapper:
                if k.lower() == 'hiddenservices':
                    for x in v:
                        for (kk, vv) in x.config_attributes():
                            yield (str(kk), str(vv))

                else:
                    # FIXME actually, is this right? don't we want ALL
                    # the values in one string?!
                    for x in v:
                        yield (str(k), str(x))

            else:
                yield (str(k), str(v))

    def create_torrc(self):
        rtn = StringIO()

        for (k, v) in self.config_args():
            rtn.write(u'%s %s\n' % (k, v))

        return rtn.getvalue() class TCPHiddenServiceEndpoint(object):
    """
    This represents something listening on an arbitrary local port
    that has a Tor configured with a Hidden Service pointing at
    it. :api:`twisted.internet.endpoints.TCP4ServerEndpoint
    <TCP4ServerEndpoint>` is used under the hood to do the local
    listening.

    There are three main ways to use this class, and you are
    encouraged to use the @classmethod ways of creating instances:
    `system_tor <#txtorcon.TCPHiddenServiceEndpoint.system_tor>`_,
    `global_tor <#txtorcon.TCPHiddenServiceEndpoint.global_tor>`_,
    and `private_tor <#txtorcon.TCPHiddenServiceEndpoint.private_tor>`_

    1. system_tor(...) connects to an already-started tor on the
       endpoint you specify; stricly speaking not a "system" tor since
       you could have spawned it some other way. See `Tor bug 11291
       <https://trac.torproject.org/projects/tor/ticket/11291>`_
       however.

    2. global_tor(...) refers to a single possible Tor instance
       per python process. So the first call to this launches a new Tor, and
       subsequent calls re-use the existing Tor (that is, add more hidden
       services to it).

    3. private_tor(...) launches a new Tor instance no matter what, so
       it will have just the one hidden serivce on it.

    If you need to set configuration options that are not reflected in
    any of the method signatures above, you'll have to construct an
    instance of this class yourself (i.e. with a TorConfig instance
    you've created).

    No matter how you came by this endpoint instance, you should call
    `listen()` on it to trigger any work required to create the
    service: Tor will be launched or connected-to; config for the
    onion service will be added; the uploading of descriptors is
    awaited.

    The ``Deferred`` from ``listen()`` will fire with an
    ``IListeningPort`` whose ``getHost()`` will return a
    :class:`txtorcon.TorOnionAddress`. The port object also has a
    `.onion_service` property which resolves to the
    :class:`txtorcon.IOnionService` or
    :class:`txtorcon.IAuthenticatedOnionClients` instance (and from
    which you can recover private keys, the hostname, etc)

    :ivar onion_uri: the public key, like ``timaq4ygg2iegci7.onion``
        which came from the hidden_service_dir's ``hostname`` file

    :ivar onion_private_key: the contents of ``hidden_service_dir/private_key``

    :ivar hidden_service_dir: the data directory, either passed in or created
        with ``tempfile.mkdtemp``

    **NOTE** that if you do not specify a `version=` then you will get
    a version 2 service (new onion APIs return version=3 services by
    default). This is for backwards-compatiblity reasons, as version=
    didn't exist before 18.0.0
    """

    @classmethod
    def system_tor(cls, reactor, control_endpoint, public_port,
                   hidden_service_dir=None,
                   local_port=None,
                   ephemeral=None,
                   auth=None,
                   private_key=None,
                   version=None,
                   single_hop=None):
        """
        This returns a TCPHiddenServiceEndpoint connected to the
        endpoint you specify in `control_endpoint`. After connecting, a
        single hidden service is added. The endpoint can be a Unix
        socket if Tor's `ControlSocket` option was used (instead of
        `ControlPort`).

        .. note::

            If Tor bug #11291 is not yet fixed, this won't work if you
            only have Group access. XXX FIXME re-test
        """

        from txtorcon.controller import connect
        tor = connect(reactor, control_endpoint)
        tor.addCallback(lambda t: t.get_config())
        # tor is a Deferred
        return TCPHiddenServiceEndpoint(
            reactor, tor, public_port,
            hidden_service_dir=hidden_service_dir,
            local_port=local_port,
            ephemeral=ephemeral,
            private_key=private_key,
            auth=auth,
            version=version,
            single_hop=single_hop,
        )

    @classmethod
    def global_tor(cls, reactor, public_port,
                   hidden_service_dir=None,
                   local_port=None,
                   control_port=None,
                   stealth_auth=None,  # backwards-compat; don't use
                   auth=None,
                   ephemeral=None,
                   private_key=None,
                   version=None,
                   single_hop=None):
        """
        This returns a TCPHiddenServiceEndpoint connected to a
        txtorcon global Tor instance. The first time you call this, a
        new Tor will be launched. Subsequent calls will re-use the
        same connection (in fact, the very same TorControlProtocol and
        TorConfig instances). If the options you pass are incompatible
        with an already-launched Tor, RuntimeError will be thrown.

        It's probably best to not specify any option besides
        `public_port`, `hidden_service_dir`, and maybe `local_port`
        unless you have a specific need to.

        You can also access this global txtorcon instance via
        :meth:`txtorcon.get_global_tor_instance` (which is precisely what
        this method uses to get it).

        All keyword options have defaults (e.g. random ports, or
        tempdirs).

        :param stealth_auth: **Deprecated**
            None, or a list of strings -- one for each stealth
            authenticator you require. Use `auth=` now.

        :param auth: None or an :class:`txtorcon.AuthBasic` or
            :class:`txtorcon.AuthStealth` instance
        """

        def progress(*args):
            progress.target(*args)
        tor = get_global_tor_instance(
            reactor,
            control_port=control_port,
            progress_updates=progress
        )
        # tor is a Deferred here, but endpoint resolves it in the
        # listen() call. Also, we want it to resolve to a TorConfig,
        # not a Tor
        tor.addCallback(lambda tor: tor.get_config())
        r = TCPHiddenServiceEndpoint(
            reactor, tor, public_port,
            hidden_service_dir=hidden_service_dir,
            local_port=local_port,
            auth=auth,
            ephemeral=ephemeral,
            private_key=private_key,
            version=version,
            single_hop=single_hop,
        )
        progress.target = r._tor_progress_update
        return r

    @classmethod
    def private_tor(cls, reactor, public_port,
                    hidden_service_dir=None,
                    local_port=None,
                    control_port=None,
                    ephemeral=None,
                    private_key=None,
                    auth=None,
                    version=None,
                    single_hop=None):
        """
        This returns a TCPHiddenServiceEndpoint that's always
        connected to its own freshly-launched Tor instance. All
        keyword options have defaults (e.g. random ports, or
        tempdirs).
        """

        def progress(*args):
            progress.target(*args)

        from .controller import launch
        tor = launch(
            reactor,
            progress_updates=progress,
            control_port=control_port,
        )
        tor.addCallback(lambda t: t.get_config())
        r = TCPHiddenServiceEndpoint(
            reactor, tor, public_port,
            hidden_service_dir=hidden_service_dir,
            local_port=local_port,
            ephemeral=ephemeral,
            private_key=private_key,
            auth=auth,
            version=version,
            single_hop=single_hop,
        )
        progress.target = r._tor_progress_update
        return r

    def __init__(self, reactor, config, public_port,
                 hidden_service_dir=None,
                 local_port=None,
                 auth=None,  # new way to pass authentication information
                 stealth_auth=None,  # deprecated; use auth=
                 ephemeral=None,  # will be set to True, unless hsdir spec'd
                 private_key=None,
                 group_readable=False,
                 version=None,
                 single_hop=None):
        """
        :param reactor:
            :api:`twisted.internet.interfaces.IReactorTCP` provider

        :param config:
            :class:`txtorcon.TorConfig` instance or a Deferred yielding one

        :param public_port:
            The port number we will advertise in the hidden serivces
            directory.

        :param local_port:
            The port number we will perform our local tcp listen on and
            receive incoming connections from the tor process.

        :param hidden_service_dir:
            If not None, point to a HiddenServiceDir directory
            (i.e. with "hostname" and "private_key" files in it). If
            not provided, one is created with temp.mkdtemp() AND
            DELETED when the reactor shuts down.

        :param auth:
            An AuthBasic or AuthStealth instance (or None)

        :param stealth_auth:
            **Deprecated; use ``auth=``**. This is for backwards-comapatibility only.

        :param endpoint_generator:
            A callable that generates a new instance of something that
            implements IServerEndpoint (by default TCP4ServerEndpoint)

        :param group_readable:
            Only for filesystem services. Causes the directory to be
            group-readable when Tor creates it.

        :param version:
            Either None, 2 or 3 to specify a version 2 service or
            Proposition 224 (version 3) service.

        :param single_hop: if True, pass the `NonAnonymous` flag. Note
            that Tor options `HiddenServiceSingleHopMode`,
            `HiddenServiceNonAnonymousMode` must be set to `1` and there
            must be no `SOCKSPort` configured for this to actually work.
        """

        # this supports API backwards-compatibility -- if you didn't
        # explicitly specify ephemeral=True, but *did* set
        # hidden_service_dir
        if ephemeral is None:
            ephemeral = True
            if hidden_service_dir is not None:
                ephemeral = False

        # backwards-compatibility for stealth_auth= kwarg
        if stealth_auth is not None:
            log.msg("'stealth_auth' is deprecated; use auth= instead")
            if auth is not None:
                raise ValueError(
                    "Both stealth_auth= and auth= passed; use auth= only for new code"
                )
            auth = AuthStealth(stealth_auth)
            stealth_auth = None

        if ephemeral and isinstance(auth, AuthStealth):
            raise ValueError(
                "'ephemeral=True' onion services don't support 'stealth' auth"
            )

        if ephemeral and hidden_service_dir is not None:
            raise ValueError(
                "Specifying 'hidden_service_dir' is incompatible"
                " with 'ephemeral=True'"
            )

        if private_key is not None and not ephemeral:
            raise ValueError(
                "'private_key' only understood for ephemeral services"
            )

        if single_hop and not ephemeral:
            raise ValueError(
                "'single_hop=' flag only makes sense for ephemeral onions"
            )

        self._reactor = reactor
        self._config = defer.maybeDeferred(lambda: config)
        self.public_port = public_port
        self.local_port = local_port
        self.auth = auth

        self.ephemeral = ephemeral
        self.private_key = private_key
        # if we're an ephemeral service, hidden_service_dir is None
        # and ephemeral is True
        self.hidden_service_dir = hidden_service_dir
        self.tcp_listening_port = None
        self.hiddenservice = None
        self.group_readable = group_readable
        self.version = version
        self.single_hop = single_hop
        self.retries = 0

        if self.version is None:
            self.version = 2

        '''for IProgressProvider to add_progress_listener'''
        self.progress_listeners = []

        if not self.ephemeral:
            if self.hidden_service_dir is None:
                self.hidden_service_dir = tempfile.mkdtemp(prefix='tortmp')
                log.msg('Will delete "%s" at shutdown.' % self.hidden_service_dir)
                delete = functools.partial(shutil.rmtree, self.hidden_service_dir)
                self._reactor.addSystemEventTrigger('before', 'shutdown', delete)

    @property
    def onion_uri(self):
        if self.hiddenservice is None:
            return None
        if IAuthenticatedOnionClients.providedBy(self.hiddenservice):
            return _maybe_unique_host(self.hiddenservice)
        return self.hiddenservice.hostname

    @property
    def onion_private_key(self):
        if self.hiddenservice is None:
            return None
        return self.hiddenservice.private_key

    def add_progress_listener(self, listener):
        """IProgressProvider API"""
        self.progress_listeners.append(listener)

    def _descriptor_progress_update(self, prog, tag, summary):
        # 'prog' here ranges from 0 -> 100.0 but we've only reserved
        # "10.0" of the range for the "upload descriptors" portion,
        # and we know that Tor has launched, so we're mapping "prog"
        # to the "100 -> 110" part of the range
        scaled_prog = (100.0 + (prog / 10.0)) * (100.0 / 110.0)
        for p in self.progress_listeners:
            try:
                p(scaled_prog, tag, summary)
            except Exception:
                log.err()

    def _tor_progress_update(self, prog, tag, summary):
        # we re-adjust the percentage-scale, using 105% and 110% for
        # the two parts of waiting for descriptor upload. That is, we
        # want: 110 * constant == 100.0
        scaled_prog = prog * (100.0 / 110.0)
        log.msg('%d%% %s' % (scaled_prog, summary))
        for p in self.progress_listeners:
            try:
                p(scaled_prog, tag, summary)
            except Exception:
                log.err()

    @defer.inlineCallbacks
    def listen(self, protocolfactory):
        """
        Implement :api:`twisted.internet.interfaces.IStreamServerEndpoint
        <IStreamServerEndpoint>`.

        Returns a Deferred that delivers an
        :api:`twisted.internet.interfaces.IListeningPort` implementation.

        This object will also have a `.onion_service` property which
        resolve to an instance implementing
        :class:`txtorcon.IOnionService` or
        :class:`txtorcon.IAuthenticatedOnionClients` (depending on
        whether the service is authenticated or not).

        At this point, Tor will have fully started up and successfully
        accepted the hidden service's config. The Onion Service's
        descriptor will be uploaded to at least one directory (as
        reported via the `HS_DESC` event).
        """

        self.protocolfactory = protocolfactory

        # self._config is always a Deferred; see __init__
        self._config = yield self._config
        if not isinstance(self._config, TorConfig):
            raise ValueError(
                'Expected a TorConfig instance but '
                'got "{}.{}" instead.'.format(
                    self._config.__class__.__module__,
                    self._config.__class__.__name__,
                )
            )
        # just to be sure:
        yield self._config.post_bootstrap

        # XXX - perhaps allow the user to pass in an endpoint
        # descriptor and make this one the default? Then would
        # probably want to check for "is a local interface or not" and
        # at *least* warn if it's not local...
        self.tcp_endpoint = serverFromString(
            self._reactor,
            'tcp:0:interface=127.0.0.1',
        )
        d = self.tcp_endpoint.listen(self.protocolfactory)
        self.tcp_listening_port = yield d
        self.local_port = self.tcp_listening_port.getHost().port

        # XXX can we detect if tor supports Unix sockets here? I guess
        # we could try "unix:/tmp/blarg", and if it fails, try
        # "tcp:0:interface=127.0.0.1" ...?

        # specifically NOT creating the hidden-service dir; letting
        # Tor do it will more-likely result in a usable situation...
        if not self.ephemeral:
            if not os.path.exists(self.hidden_service_dir):
                log.msg(
                    'Noting that "%s" does not exist; letting Tor create it.' %
                    self.hidden_service_dir
                )

        # see note in _tor_progress_update; we extend the percent
        # range to 110% for the descriptor upload
        self._descriptor_progress_update(
            0.0, 'wait_descriptor', 'uploading descriptor'
        )

        # see if the hidden-serivce instance we want is already in the
        # config; for non-ephemeral services, the directory is unique;
        # for ephemeral services, the key should exist and be unique.
        already = False
        if self.ephemeral:
            already = self.hiddenservice is not None
        else:
            hs_dirs = [hs.dir for hs in self._config.HiddenServices if hasattr(hs, 'dir')]
            already = os.path.abspath(self.hidden_service_dir) in hs_dirs

        if not already:
            if self.ephemeral:
                if self.auth is not None:
                    create_d = EphemeralAuthenticatedOnionService.create(
                        self._reactor,
                        self._config,
                        ['%d 127.0.0.1:%d' % (self.public_port, self.local_port)],
                        private_key=self.private_key,
                        detach=False,
                        progress=self._descriptor_progress_update,
                        version=self.version,
                        auth=self.auth,
                        single_hop=self.single_hop,
                    )

                else:
                    create_d = EphemeralOnionService.create(
                        self._reactor,
                        self._config,
                        ['%d 127.0.0.1:%d' % (self.public_port, self.local_port)],
                        private_key=self.private_key,
                        detach=False,
                        progress=self._descriptor_progress_update,
                        version=self.version,
                        single_hop=self.single_hop,
                    )
            else:
                if self.auth is not None:
                    create_d = FilesystemAuthenticatedOnionService.create(
                        self._reactor,
                        self._config,
                        self.hidden_service_dir,
                        ['%d 127.0.0.1:%d' % (self.public_port, self.local_port)],
                        auth=self.auth,  # AuthBasic or AuthStealth
                        progress=self._descriptor_progress_update,
                        group_readable=self.group_readable,
                        version=self.version,
                    )
                else:
                    create_d = FilesystemOnionService.create(
                        self._reactor,
                        self._config,
                        self.hidden_service_dir,
                        ['%d 127.0.0.1:%d' % (self.public_port, self.local_port)],
                        progress=self._descriptor_progress_update,
                        group_readable=self.group_readable,
                        version=self.version,
                    )
            self.hiddenservice = yield create_d

        else:
            if not self.ephemeral:
                for hs in self._config.HiddenServices:
                    if hs.dir == os.path.abspath(self.hidden_service_dir):
                        self.hiddenservice = hs

        assert self.hiddenservice is not None, "internal error"

        if IAuthenticatedOnionClients.providedBy(self.hiddenservice):
            log.msg('Started authenticated onion service on:')
            for nm in self.hiddenservice.client_names():
                log.msg('  {}: {}'.format(nm, self.hiddenservice.get_client(nm).hostname))
        else:
            log.msg('Started onion service on %s:%d' % (self.onion_uri, self.public_port))

        # XXX should just return self.hiddenservice here??
        # -> no, don't think so for a couple reasons:

        # 1. it's "ports" in the services, so "TorOnionListeningPort"
        # is the only thing that knows there's just one in this one
        # (so can provide .local_port -> shoujld be local_endpoint I
        # guess actually...)
        # 2. anyway, can provide access to the "real" hs anyway if we want
        defer.returnValue(
            TorOnionListeningPort(
                self.tcp_listening_port,
                self.public_port,
                self.hiddenservice,
                self._config,
            )
        ) class TorClientEndpoint(object):
    """
    An IStreamClientEndpoint which establishes a connection via Tor.

    You should not instantiate these directly; use
    ``clientFromString()``, :meth:`txtorcon.Tor.stream_via` or
    :meth:`txtorcon.Circuit.stream_via`

    :param host:
        The hostname to connect to. This of course can be a Tor Hidden
        Service onion address.

    :param port: The tcp port or Tor Hidden Service port.

    :param socks_endpoint: An IStreamClientEndpoint pointing at (one
        of) our Tor's SOCKS ports. These can be instantiated with
        :meth:`txtorcon.TorConfig.socks_endpoint`.

    :param tls: Can be False or True (to get default Browser-like
        hostname verification) or the result of calling
        optionsForClientTLS() yourself. Default is True.
    """

    socks_ports_to_try = [9050, 9150]

    @classmethod
    def from_connection(cls, reactor, control_protocol, host, port,
                        tls=None,
                        socks_endpoint=None):
        if socks_endpoint is None:
            socks_endpoint = _create_socks_endpoint(reactor, control_protocol)
        return TorClientEndpoint(
            host, port,
            socks_endpoint=socks_endpoint,
            tls=tls,
            reactor=reactor,
        )

    def __init__(self,
                 host, port,
                 socks_endpoint=None,  # can be Deferred
                 tls=False,

                 # XXX our custom SOCKS stuff doesn't support auth (yet?)
                 socks_username=None, socks_password=None,
                 reactor=None, **kw):
        if host is None or port is None:
            raise ValueError('host and port must be specified')

        self.host = host
        self.port = int(port)
        self._socks_endpoint = socks_endpoint
        self._socks_username = socks_username
        self._socks_password = socks_password
        self._tls = tls
        # XXX FIXME we 'should' probably include 'reactor' as the
        # first arg to this class, but technically that's a
        # breaking change :(
        self._reactor = reactor
        if reactor is None:
            from twisted.internet import reactor
            self._reactor = reactor

        # backwards-compatibility: you used to specify a TCP SOCKS
        # endpoint via socks_host= and socks_port= kwargs
        if self._socks_endpoint is None:
            try:
                self._socks_endpoint = TCP4ClientEndpoint(
                    reactor,
                    kw['socks_hostname'],
                    kw['socks_port'],
                )
                # XXX should deprecation-warn here
            except KeyError:
                pass

        # this is a separate "if" from above in case socks_endpoint
        # was None but the user specified the (old)
        # socks_hostname/socks_port (in which case we do NOT want
        # guessing_enabled
        if self._socks_endpoint is None:
            self._socks_guessing_enabled = True
        else:
            self._socks_guessing_enabled = False

        # XXX think, do we want to expose these like this? Or some
        # other way (because they're for stream-isolation, not actual
        # auth)
        self._socks_username = socks_username
        self._socks_password = socks_password
        self._when_address = SingleObserver()

    def _get_address(self):
        """
        internal helper.

        *le sigh*. This is basically just to support
        TorCircuitEndpoint; see TorSocksEndpoint._get_address(). There
        shouldn't be any need for "actual users" to need this!

        This returns a Deferred that fires once:
          - we have an underlying SOCKS5 endpoint
          - ...and it has received a local connection (and hence the address/port)
        """
        return self._when_address.when_fired()

    @defer.inlineCallbacks
    def connect(self, protocolfactory):
        last_error = None
        # XXX fix in socks.py stuff for socks_username, socks_password
        if self._socks_username or self._socks_password:
            raise RuntimeError(
                "txtorcon socks support doesn't yet do username/password"
            )
        if self._socks_endpoint is not None:
            socks_ep = TorSocksEndpoint(
                self._socks_endpoint,
                self.host, self.port,
                self._tls,
            )
            # forward the address to any listeners we have
            socks_ep._get_address().addCallback(self._when_address.fire)
            proto = yield socks_ep.connect(protocolfactory)
            defer.returnValue(proto)
        else:
            for socks_port in self.socks_ports_to_try:
                tor_ep = TCP4ClientEndpoint(
                    self._reactor,
                    "127.0.0.1",  # XXX socks_hostname, no?
                    socks_port,
                )
                socks_ep = TorSocksEndpoint(tor_ep, self.host, self.port, self._tls)
                # forward the address to any listeners we have
                socks_ep._get_address().addCallback(self._when_address.fire)
                try:
                    proto = yield socks_ep.connect(protocolfactory)
                    defer.returnValue(proto)

                except error.ConnectError as e0:
                    last_error = e0
            if last_error is not None:
                raise last_error class IProgressProvider(Interface):
    """FIXME move elsewhere? think harder?"""
    def add_progress_listener(listener):
        """
        Adds a progress listener. The listener is a callable that gets
        called with 3 arguments corresponding to Tor's updates:
        (percent, tag, message). percent is an integer from 0 to 100,
        tag and message are both strings. (message is the
        human-readable one)
        """ class TorOnionAddress(FancyEqMixin, object):
    """
    A ``TorOnionAddress`` represents the public address of a Tor onion
    service. Instances of these come from calling the Twisted method
    `.getHost()` on :api:`twisted.internet.interfaces.IListeningPort`
    which was returned from the :class:`txtorcon.TCPHiddenServiceEndpoint.listen`

    :ivar type: A string describing the type of transport, 'onion'.

    :ivar onion_port: The public port we're advertising

    :ivar onion_key: the private key for the service
    """
    compareAttributes = ('type', 'onion_port', 'onion_key')
    type = 'onion'

    # for authenticated services, there is a private-key for "the
    # service"; for stealth-auth'd services, there are *also*
    # private-keys for each client

    def __init__(self, port, hs):
        self.onion_port = port

        # this gets a bit weird .. partially for backwards-
        # compatibility: .onion_uri is/was an existing property -- but
        # doesn't always make sense. so, users should be encouraged to
        # use .onion_service and access things directly (i.e. they
        # know if they have an authenticated service or not, or can
        # find out via .providedBy())

        if IAuthenticatedOnionClients.providedBy(hs):
            try:
                self.onion_uri = _maybe_unique_host(hs)
            except ValueError:
                self.onion_uri = None
        else:
            self.onion_uri = hs.hostname
        self._hiddenservice = hs

    @property
    def onion_service(self):
        return self._hiddenservice

    @property
    def onion_key(self):
        return self._hiddenservice.private_key

    def __repr__(self):
        return '%s(%s)' % (self.__class__.__name__, self.onion_uri)

    def __hash__(self):
        # should be "URIs", not URI
        return hash((self.type, self.onion_uri, self.onion_port)) class IAuthenticatedOnionClients(Interface):
    """
    This encapsulates both 'stealth' and 'basic' authenticated Onion
    services, whether ephemeral or not.

    Each client has an arbitrary (ASCII, no spaces) name. You may
    access the clients with `get_client`, which will all be
    :class:`txtorcon.IOnionClient` instances.
    """

    def get_permanent_id(self):
        """
        :return: the service's permanent id, in hex

        (For authenticated services, this is not the same as the
        .onion URI of any of the clients). The Permanent ID is the
        base32 encoding of the first 10 bytes of the SHA1 hash of the
        public-key of the service.
        """

    def client_names(self):
        """
        :return: list of str instances, one for each client
        """

    def get_client(self, name):
        """
        :return: object implementing IOnionClient for the named client
        """

    def add_client(self, name):
        """
        probably should return a Deferred?
        """

    def del_client(self, name):
        """
        probably should return a Deferred?
        """ class IOnionService(Interface):
    """
    Encapsulates a single, ephemeral onion service.

    If this instance happens to be a filesystem-based service (instead
    of ephemeral), it shall implement IFilesystemOnionService as well
    (which is a subclass of this).

    If this object happens to represent an authenticated service, it
    shall implement IAuthenticatedOnionClients ONLY (not this
    interface too; IAuthenticatedOnionClients returns *lists* of
    IOnionClient instances which are a subclass of
    IOnionService; see :class:`txtorcon.IAuthenticatedOnionClients`).

    For non-authenticated services, there will be one of these per
    directory (i.e. HiddenServiceDir) if using non-ephemeral services,
    or one per ADD_ONION for ephemeral hidden services.

    For authenticated services, there is an instance implementing this
    interface for each "client" of the authenticated service. In the
    "basic" case, the .onion URI happens to be the same for each one
    (with a different authethentication token) whereas for a "stealth"
    sevice the .onion URI is different.
    """
    hostname = Attribute("hostname, including .onion")  # XXX *with* .onion? or not?
    private_key = Attribute("Private key blob (bytes)")
    ports = Attribute("list of str; the ports lines like 'public_port host:local_port'") def _compute_permanent_id(private_key):
    """
    Internal helper. Return an authenticated service's permanent ID
    given an RSA private key object.

    The permanent ID is the base32 encoding of the SHA1 hash of the
    first 10 bytes (80 bits) of the public key.
    """
    pub = private_key.public_key()
    p = pub.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.PKCS1
    )
    z = ''.join(p.decode('ascii').strip().split('\n')[1:-1])
    b = base64.b64decode(z)
    h1 = hashlib.new('sha1')
    h1.update(b)
    permanent_id = h1.digest()[:10]
    return base64.b32encode(permanent_id).lower().decode('ascii') class AuthStealth(_AuthCommon):
    """
    Authentication details for 'stealth' auth.
    """
    auth_type = 'stealth'
    # note that _AuthCommon.__init__ takes 'clients' class AuthBasic(_AuthCommon):
    """
    Authentication details for 'basic' auth.
    """
    auth_type = 'basic'
    # note that _AuthCommon.__init__ takes 'clients' class NoOpProtocolFactory:
    """
    This is an IProtocolFactory that does nothing. Used for testing,
    and for :method:`available_tcp_port`
    """
    def noop(self, *args, **kw):
        pass
    buildProtocol = noop
    doStart = noop
    doStop = noop class SingleObserver(object):
    """
    A helper for ".when_*()" sort of functions.
    """
    _NotFired = object()

    def __init__(self):
        self._observers = []
        self._fired = self._NotFired

    def when_fired(self):
        d = defer.Deferred()
        if self._fired is not self._NotFired:
            d.callback(self._fired)
        else:
            self._observers.append(d)
        return d

    def fire(self, value):
        if self._observers is None:
            return  # raise RuntimeError("already fired") ?
        self._fired = value
        for d in self._observers:
            d.callback(self._fired)
        self._observers = None
        return value  # so we're transparent if used as a callback @deprecated(_Version("txtorcon", 18, 0, 0))
@defer.inlineCallbacks
def get_global_tor(reactor, control_port=None,
                   progress_updates=None,
                   _tor_launcher=None):
    """
    See description of :class:`txtorcon.TCPHiddenServiceEndpoint`'s
    class-method ``global_tor``

    :param control_port:
        a TCP port upon which to run the launched Tor's
        control-protocol (selected by the OS by default).

    :param progress_updates:
        A callable that takes 3 args: ``percent, tag, message`` which
        is called when Tor announcing some progress setting itself up.

    :returns:
        a ``Deferred`` that fires a :class:`txtorcon.TorConfig` which is
        bootstrapped.

    The _tor_launcher keyword arg is internal-only.
    """
    tor = yield get_global_tor_instance(
        reactor,
        control_port=control_port,
        progress_updates=progress_updates,
        _tor_launcher=_tor_launcher,
    )
    cfg = yield tor.get_config()
    defer.returnValue(cfg) @defer.inlineCallbacks
def _create_socks_endpoint(reactor, control_protocol, socks_config=None):
    """
    Internal helper.

    This uses an already-configured SOCKS endpoint from the attached
    Tor, or creates a new TCP one (and configures Tor with it). If
    socks_config is non-None, it is a SOCKSPort line and will either
    be used if it already exists or will be created.
    """
    socks_ports = yield control_protocol.get_conf('SOCKSPort')
    if socks_ports:
        socks_ports = list(socks_ports.values())[0]
        if not isinstance(socks_ports, list):
            socks_ports = [socks_ports]
        # see TorConfig for more fun-times regarding *PortLines, including
        # the __*Port things...
        if socks_ports == ['DEFAULT']:
            default = yield control_protocol.get_conf_single('__SocksPort')
            socks_ports = [default]
    else:
        # return from get_conf was an empty dict; we want a list
        socks_ports = []

    # everything in the SocksPort list can include "options" after the
    # initial value. We don't care about those, but do need to strip
    # them.
    socks_ports = [port.split()[0] for port in socks_ports]

    # could check platform? but why would you have unix ports on a
    # platform that doesn't?
    unix_ports = set([p for p in socks_ports if p.startswith('unix:')])
    tcp_ports = set(socks_ports) - unix_ports

    socks_endpoint = None
    for p in list(unix_ports) + list(tcp_ports):  # prefer unix-ports
        if socks_config and p != socks_config:
            continue
        try:
            socks_endpoint = _endpoint_from_socksport_line(reactor, p)
        except Exception as e:
            log.err(
                Failure(),
                "failed to process SOCKS port '{}': {}".format(p, e)
            )

    # if we still don't have an endpoint, nothing worked (or there
    # were no SOCKSPort lines at all) so we add config to tor
    if socks_endpoint is None:
        if socks_config is None:
            # is a unix-socket in /tmp on a supported platform better than
            # this?
            port = yield available_tcp_port(reactor)
            socks_config = str(port)
        socks_ports.append(socks_config)

        # NOTE! We must set all the ports in one command or we'll
        # destroy pre-existing config
        args = []
        for p in socks_ports:
            args.append('SOCKSPort')
            args.append(p)
        yield control_protocol.set_conf(*args)
        socks_endpoint = _endpoint_from_socksport_line(reactor, socks_config)

    assert socks_endpoint is not None
    defer.returnValue(socks_endpoint) class TorCircuitEndpoint(object):
    def __init__(self, reactor, torstate, circuit, target_endpoint):
        self._reactor = reactor
        self._state = torstate
        self._target_endpoint = target_endpoint  # a TorClientEndpoint
        self._circuit = circuit

    @defer.inlineCallbacks
    def connect(self, protocol_factory):
        """IStreamClientEndpoint API"""
        # need to:
        # 1. add 'our' attacher to state
        # 2. do the "underlying" connect
        # 3. recognize our stream
        # 4. attach it to our circuit

        attacher = yield _get_circuit_attacher(self._reactor, self._state)
        # note that we'll only ever add an attacher once, and then it
        # stays there "forever". so if you never call the .stream_via
        # or .web_agent APIs, set_attacher won't get called .. but if
        # you *do*, then you can't call set_attacher yourself (because
        # that's an error). See discussion in set_attacher on
        # TorState or issue #169

        yield self._circuit.when_built()
        connect_d = self._target_endpoint.connect(protocol_factory)
        attached_d = attacher.add_endpoint(self._target_endpoint, self._circuit)
        proto = yield connect_d
        yield attached_d
        defer.returnValue(proto) @defer.inlineCallbacks
def _get_circuit_attacher(reactor, state):
    if _get_circuit_attacher.attacher is None:
        _get_circuit_attacher.attacher = _CircuitAttacher()
        yield state.set_attacher(_get_circuit_attacher.attacher, reactor)
    defer.returnValue(_get_circuit_attacher.attacher) class Tor(object):
    """
    I represent a single instance of Tor and act as a Builder/Factory
    for several useful objects you will probably want. There are two
    ways to create a Tor instance:

       - :func:`txtorcon.connect` to connect to a Tor that is already
         running (e.g. Tor Browser Bundle, a system Tor, ...).
       - :func:`txtorcon.launch` to launch a fresh Tor instance

    The stable API provided by this class is :class:`txtorcon.interface.ITor`

    If you desire more control, there are "lower level" APIs which are
    the very ones used by this class. However, this "highest level"
    API should cover many use-cases::

        import txtorcon

        @inlineCallbacks
        def main(reactor):
            # tor = yield txtorcon.connect(UNIXClientEndpoint(reactor, "/var/run/tor/control"))
            tor = yield txtorcon.launch(reactor)

            onion_ep = tor.create_onion_endpoint(port=80)
            port = yield onion_ep.listen(Site())
            print(port.getHost())
    """

    def __init__(self, reactor, control_protocol, _tor_config=None, _process_proto=None, _non_anonymous=None):
        """
        don't instantiate this class yourself -- instead use the factory
        methods :func:`txtorcon.launch` or :func:`txtorcon.connect`
        """
        self._protocol = control_protocol
        self._config = _tor_config
        self._reactor = reactor
        # this only passed/set when we launch()
        self._process_protocol = _process_proto
        # cache our preferred socks port (please use
        # self._default_socks_endpoint() to get one)
        self._socks_endpoint = None
        # True if we've turned on non-anonymous mode / Onion services
        self._non_anonymous = _non_anonymous

    @inlineCallbacks
    def quit(self):
        """
        Closes the control connection, and if we launched this Tor
        instance we'll send it a TERM and wait until it exits.
        """
        if self._protocol is not None:
            yield self._protocol.quit()
        if self._process_protocol is not None:
            yield self._process_protocol.quit()
        if self._protocol is None and self._process_protocol is None:
            raise RuntimeError(
                "This Tor has no protocol instance; we can't quit"
            )
        if self._protocol is not None:
            yield self._protocol.on_disconnect

    @property
    def process(self):
        """
        An object implementing
        :api:`twisted.internet.interfaces.IProcessProtocol` if this
        Tor instance was launched, or None.
        """
        if self._process_protocol:
            return self._process_protocol
        return None

    @property
    def protocol(self):
        """
        The TorControlProtocol instance that is communicating with this
        Tor instance.
        """
        return self._protocol

    @property
    def version(self):
        return self._protocol.version

    @inlineCallbacks
    def get_config(self):
        """
        :return: a Deferred that fires with a TorConfig instance. This
            instance represents up-to-date configuration of the tor
            instance (even if another controller is connected). If you
            call this more than once you'll get the same TorConfig back.
        """
        if self._config is None:
            self._config = yield TorConfig.from_protocol(self._protocol)
        returnValue(self._config)

    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=``)
        """
        if self._non_anonymous:
            raise Exception(
                "Cannot use web_agent when in non_anonymous mode"
            )
        # 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,
        )

    @inlineCallbacks
    def dns_resolve(self, hostname):
        """
        :param hostname: a string

        :returns: a Deferred that calbacks with the hostname as looked-up
            via Tor (or errback).  This uses Tor's custom extension to the
            SOCKS5 protocol.
        """
        socks_ep = yield self._default_socks_endpoint()
        ans = yield socks.resolve(socks_ep, hostname)
        returnValue(ans)

    @inlineCallbacks
    def dns_resolve_ptr(self, ip):
        """
        :param ip: a string, like "127.0.0.1"

        :returns: a Deferred that calbacks with the IP address as
            looked-up via Tor (or errback).  This uses Tor's custom
            extension to the SOCKS5 protocol.
        """
        socks_ep = yield self._default_socks_endpoint()
        ans = yield socks.resolve_ptr(socks_ep, ip)
        returnValue(ans)

    @inlineCallbacks
    def add_onion_authentication(self, onion_host, token):
        """
        Add a client-side authentication token for a particular Onion
        service.
        """
        # if we add the same onion twice, Tor rejects us. We throw an
        # error if we already have that .onion but the incoming token
        # doesn't match
        if isinstance(onion_host, bytes):
            onion_host = onion_host.decode('ascii')

        config = yield self.get_config()
        tokens = {
            servauth.split()[0]: servauth.split()[1]
            for servauth in config.HidServAuth
        }
        try:
            maybe_token = tokens[onion_host]
            if maybe_token != token:
                raise ValueError(
                    "Token conflict for host '{}'".format(onion_host)
                )
            return
        except KeyError:
            pass

        # add our onion + token combo
        config.HidServAuth.append(
            u"{} {}".format(onion_host, token)
        )
        yield config.save()

    @inlineCallbacks
    def remove_onion_authentication(self, onion_host):
        """
        Remove a token for an onion host

        :returns: True if successful, False if there wasn't a token
            for that host.
        """
        if isinstance(onion_host, bytes):
            onion_host = onion_host.decode('ascii')

        config = yield self.get_config()
        to_remove = None
        for auth in config.HidServAuth:
            host, token = auth.split()
            if host == onion_host:
                to_remove = auth

        if to_remove is not None:
            config.HidServAuth.remove(to_remove)
            yield config.save()
            returnValue(True)
        returnValue(False)

    def onion_authentication(self, onion_host, token):
        """
        (Python3 only!) This returns an async context-manager that will
        add and remove onion authentication. For example, inside an
        `async def` method that's had `ensureDeferred` called on it::

            async with tor.onion_authentication("timaq4ygg2iegci7.onion", "seekrit token"):
                agent = tor.web_agent()
                resp = await agent.request(b'GET', "http://timaq4ygg2iegci7.onion/")
                body = await readBody(resp)
            # after the "async with" the token will be removed from Tor's configuration

        Under the hood, this just uses the add_onion_authentication
        and remove_onion_authentication methods so on Python2 you can
        use those together with try/finally to get the same effect.
        """
        if not HAVE_ASYNC:
            raise RuntimeError(
                "async context-managers not supported in Python3.4 or lower"
            )
        return _AsyncOnionAuthContext(
            self, onion_host, token
        )

    def stream_via(self, host, port, tls=False, socks_endpoint=None):
        """
        This returns an IStreamClientEndpoint_ instance that will use this
        Tor (via SOCKS) to visit the ``(host, port)`` indicated.

        :param host: The host to connect to. You MUST pass host-names
            to this. If you absolutely know that you've not leaked DNS
            (e.g. you save IPs in your app's configuration or similar)
            then you can pass an IP.

        :param port: Port to connect to.

        :param tls: If True, it will wrap the return endpoint in one
            that does TLS (default: False).

        :param socks_endpoint: Normally not needed (default: None)
            but you can pass an IStreamClientEndpoint_ directed at one
            of the local Tor's SOCKS5 ports (e.g. created with
            :meth:`txtorcon.TorConfig.create_socks_endpoint`). Can be
            a Deferred.

        .. _IStreamClientEndpoint: https://twistedmatrix.com/documents/current/api/twisted.internet.interfaces.IStreamClientEndpoint.html
        """
        if _is_non_public_numeric_address(host):
            raise ValueError("'{}' isn't going to work over Tor".format(host))

        if socks_endpoint is None:
            socks_endpoint = self._default_socks_endpoint()
        # socks_endpoint may be a a Deferred, but TorClientEndpoint handles it
        return TorClientEndpoint(
            host, port,
            socks_endpoint=socks_endpoint,
            tls=tls,
            reactor=self._reactor,
        )

    def create_authenticated_onion_endpoint(self, port, auth, private_key=None, version=None):
        """
        WARNING: API subject to change

        When creating an authenticated Onion service a token is
        created for each user. For 'stealth' authentication, the
        hostname is also different for each user. The difference between
        this method and :meth:`txtorcon.Tor.create_onion_endpoint` is
        in this case the "onion_service" instance implements
        :class:`txtorcon.IAuthenticatedOnionClients`.

        :returns: an object that implements IStreamServerEndpoint,
            which will create an "ephemeral" Onion service when
            ``.listen()`` is called. This uses the ``ADD_ONION`` Tor
            control-protocol command. The object returned from
            ``.listen()`` will be a :class:TorOnionListeningPort``;
            its ``.onion_service`` attribute will be a
            :class:`txtorcon.IAuthenticatedOnionClients` instance.

        :param port: the port to listen publically on the Tor network
           on (e.g. 80 for a Web server)

        :param private_key: if not None (the default), this should be
            the same blob of key material that you received from the
            :class:`txtorcon.IOnionService` object during a previous
            run (i.e. from the ``.provate_key`` attribute).

        :param version: if not None, a specific version of service to
            use; version=3 is Proposition 224 and version=2 is the
            older 1024-bit key based implementation.

        :param auth: a AuthBasic or AuthStealth instance
        """
        return TCPHiddenServiceEndpoint(
            self._reactor, self.get_config(), port,
            hidden_service_dir=None,
            local_port=None,
            ephemeral=True,
            private_key=private_key,
            version=version,
            auth=auth,
        )

    def create_onion_endpoint(self, port, private_key=None, version=None, single_hop=None):
        """
        WARNING: API subject to change

        :returns: an object that implements IStreamServerEndpoint,
            which will create an "ephemeral" Onion service when
            ``.listen()`` is called. This uses the ``ADD_ONION`` tor
            control-protocol command. The object returned from
            ``.listen()`` will be a :class:TorOnionListeningPort``;
            its ``.onion_service`` attribute will be a
            :class:`txtorcon.IOnionService` instance.

        :param port: the port to listen publically on the Tor network
           on (e.g. 80 for a Web server)

        :param private_key: if not None (the default), this should be
            the same blob of key material that you received from the
            :class:`txtorcon.IOnionService` object during a previous
            run (i.e. from the ``.private_key`` attribute).

        :param version: if not None, a specific version of service to
            use; version=3 is Proposition 224 and version=2 is the
            older 1024-bit key based implementation.

        :param single_hop: if True, pass the `NonAnonymous` flag. Note
            that Tor options `HiddenServiceSingleHopMode`,
            `HiddenServiceNonAnonymousMode` must be set to `1` and there
            must be no `SOCKSPort` configured for this to actually work.
        """
        # note, we're just depending on this being The Ultimate
        # Everything endpoint. Which seems fine, because "normal"
        # users should use this or another factory-method to
        # instantiate them...
        return TCPHiddenServiceEndpoint(
            self._reactor, self.get_config(), port,
            hidden_service_dir=None,
            local_port=None,
            ephemeral=True,
            private_key=private_key,
            version=version,
            auth=None,
            single_hop=single_hop,
        )

    def create_filesystem_onion_endpoint(self, port, hs_dir, group_readable=False, version=None):
        """
        WARNING: API subject to change

        :returns: an object that implements IStreamServerEndpoint. When
            the ``.listen()`` method is called, the endpoint will create
            an Onion service whose keys are on disk when ``.listen()`` is
            called. The object returned from ``.listen()`` will be a
            :class:TorOnionListeningPort``; its ``.onion_service``
            attribute will be a :class:`txtorcon.IOnionService` instance.

        :param port: the port to listen publically on the Tor network
           on (e.g. 80 for a Web server)

        :param hs_dir: the directory in which keys are stored for this
            service.

        :param group_readable: controls the Tor
            `HiddenServiceDirGroupReadable` which will either set (or not)
            group read-permissions on the hs_dir.

        :param version: if not None, a specific version of service to
            use; version=3 is Proposition 224 and version=2 is the
            older 1024-bit key based implementation. The default is version 3.
        """
        return TCPHiddenServiceEndpoint(
            self._reactor, self.get_config(), port,
            hidden_service_dir=hs_dir,
            local_port=None,
            ephemeral=False,
            private_key=None,
            group_readable=int(group_readable),
            version=version,
            auth=None,
        )

    def create_filesystem_authenticated_onion_endpoint(self, port, hs_dir, auth, group_readable=False, version=None):
        """
        WARNING: API subject to change

        :returns: an object that implements IStreamServerEndpoint. When
            the ``.listen()`` method is called, the endpoint will create
            an Onion service whose keys are on disk when ``.listen()`` is
            called. The object returned from ``.listen()`` will be a
            :class:TorOnionListeningPort``; its ``.onion_service``
            attribute will be a :class:`txtorcon.IOnionService` instance.

        :param port: the port to listen publically on the Tor network
           on (e.g. 80 for a Web server)

        :param hs_dir: the directory in which keys are stored for this
            service.

        :param auth: instance of :class:`txtorcon.AuthBasic` or
            :class:`txtorcon.AuthStealth` controlling the type of
            authentication to use.

        :param group_readable: controls the Tor
            `HiddenServiceDirGroupReadable` which will either set (or not)
            group read-permissions on the hs_dir.

        :param version: if not None, a specific version of service to
            use; version=3 is Proposition 224 and version=2 is the
            older 1024-bit key based implementation. The default is version 3.
        """
        return TCPHiddenServiceEndpoint(
            self._reactor, self.get_config(), port,
            hidden_service_dir=hs_dir,
            local_port=None,
            ephemeral=False,
            private_key=None,
            group_readable=int(group_readable),
            version=version,
            auth=auth,
        )

    # XXX or get_state()? and make there be always 0 or 1 states; cf. convo w/ Warner
    @inlineCallbacks
    def create_state(self):
        """
        returns a Deferred that fires with a ready-to-go
        :class:`txtorcon.TorState` instance.
        """
        state = TorState(self.protocol)
        yield state.post_bootstrap
        returnValue(state)

    def __str__(self):
        return "<Tor version='{tor_version}'>".format(
            tor_version=self._protocol.version,
        )

    @inlineCallbacks
    def is_ready(self):
        """
        :return: a Deferred that fires with True if this Tor is
            non-dormant and ready to go. This will return True if `GETINFO
            dormant` is false or if `GETINFO status/enough-dir-info` is
            true or if `GETINFO status/circuit-established` true.
        """
        info = yield self.protocol.get_info(
            "dormant",
            "status/enough-dir-info",
            "status/circuit-established",
        )
        returnValue(
            not(
                int(info["dormant"]) or
                not int(info["status/enough-dir-info"]) or
                not int(info["status/circuit-established"])
            )
        )

    @inlineCallbacks
    def become_ready(self):
        """
        Make sure Tor is no longer dormant.

        If Tor is currently dormant, it is woken up by doing a DNS
        request for torproject.org
        """
        ready = yield self.is_ready()
        if not ready:
            yield self.dns_resolve(u'torproject.org')
        return

    @inlineCallbacks
    def _default_socks_endpoint(self):
        """
        Returns a Deferred that fires with our default SOCKS endpoint
        (which might mean setting one up in our attacked Tor if it
        doesn't have one)
        """
        if self._non_anonymous:
            raise Exception(
                "Cannot use SOCKS when in non_anonymous mode"
            )
        if self._socks_endpoint is None:
            self._socks_endpoint = yield _create_socks_endpoint(self._reactor, self._protocol)
        returnValue(self._socks_endpoint)

    # For all these create_*() methods, instead of magically computing
    # the class-name from arguments (e.g. we could decide "it's a
    # Filesystem thing" if "hidden_service_dir=" is passed) we have an
    # explicit method for each type of service. This means each method
    # always returns the same type of object (good!) and user-code is
    # more explicit about what they want (also good!) .. but the
    # method names are kind of long (not-ideal)

    @inlineCallbacks
    def create_onion_service(self, ports, private_key=None, version=3,
                             progress=None, await_all_uploads=False,
                             single_hop=None, detach=None):
        """
        Create a new Onion service

        This method will create a new Onion service, returning (via
        Deferred) an instance that implements IOnionService. (To
        create authenticated onion services, see XXX). This method
        awaits at least one upload of the Onion service's 'descriptor'
        to the Tor network -- this can take from 30s to a couple
        minutes.

        :param private_key: None, ``txtorcon.DISCARD`` or a key-blob
            retained from a prior run

            Passing ``None`` means a new one will be created. It can be
            retrieved from the ``.private_key`` property of the returned
            object. You **must** retain this key yourself (and pass it in
            to this method in the future) if you wish to keep the same
            ``.onion`` domain when re-starting your program.

            Passing ``txtorcon.DISCARD`` means txtorcon will never learn the
            private key from Tor and so there will be no way to re-create
            an Onion Service on the same address after Tor exits.

        :param version: The latest Tor releases support 'Proposition
            224' (version 3) services. These are the default.

        :param progress: if provided, a function that takes 3
            arguments: ``(percent_done, tag, description)`` which may
            be called any number of times to indicate some progress has
            been made.

        :param await_all_uploads: if False (the default) then we wait
            until at least one upload of our Descriptor to a Directory
            Authority has completed; if True we wait until all have
            completed.

        :param single_hop: if True, pass the `NonAnonymous` flag. Note
            that Tor options `HiddenServiceSingleHopMode`,
            `HiddenServiceNonAnonymousMode` must be set to `1` and there
            must be no `SOCKSPort` configured for this to actually work.

        :param detach: if True, the created service won't be tied to
            this control connection and will still be active when this
            control-connection goes away (this means the service will
            appear in `GETINFO onions/detached` to all other
            controllers)
        """
        if version not in (2, 3):
            raise ValueError(
                "The only valid Onion service versions are 2 or 3"
            )
        if not isinstance(ports, Sequence) or isinstance(ports, six.string_types):
            raise ValueError("'ports' must be a sequence (list, tuple, ..)")

        processed_ports = yield _validate_ports(self._reactor, ports)
        config = yield self.get_config()
        service = yield EphemeralOnionService.create(
            reactor=self._reactor,
            config=config,
            ports=processed_ports,
            private_key=private_key,
            version=version,
            progress=progress,
            await_all_uploads=await_all_uploads,
            single_hop=single_hop,
            detach=detach,
        )
        returnValue(service)

    @inlineCallbacks
    def create_filesystem_onion_service(self, ports, onion_service_dir,
                                        version=3,
                                        group_readable=False,
                                        progress=None,
                                        await_all_uploads=False):
        """Create a new Onion service stored on disk

        This method will create a new Onion service, returning (via
        Deferred) an instance that implements IOnionService. (To
        create authenticated onion services, see XXX). This method
        awaits at least one upload of the Onion service's 'descriptor'
        to the Tor network -- this can take from 30s to a couple
        minutes.

        :param ports: a collection of ports to advertise; these are
            forwarded locally on a random port. Each entry may instead be
            a 2-tuple, which chooses an explicit local port.

        :param onion_service_dir: a path to an Onion Service
            directory.

            Tor will write a ``hostname`` file in this directory along
            with the private keys for the service (if they do not already
            exist). You do not need to retain the private key yourself.

        :param version: which kind of Onion Service to create. The
            default is ``3`` which are the Proposition 224
            services. Version ``2`` are the previous services. There are
            no other valid versions currently.

        :param group_readable: if True, Tor creates the directory with
           group read permissions. The default is False.

        :param progress: if provided, a function that takes 3
            arguments: ``(percent_done, tag, description)`` which may
            be called any number of times to indicate some progress has
            been made.

        """
        if not isinstance(ports, Sequence) or isinstance(ports, six.string_types):
            raise ValueError("'ports' must be a sequence (list, tuple, ..)")
        processed_ports = yield _validate_ports(self._reactor, ports)

        if version not in (2, 3):
            raise ValueError(
                "The only valid Onion service versions are 2 or 3"
            )
        config = yield self.get_config()
        service = yield FilesystemOnionService.create(
            reactor=self._reactor,
            config=config,
            hsdir=onion_service_dir,
            ports=processed_ports,
            version=version,
            group_readable=group_readable,
            progress=progress,
            await_all_uploads=await_all_uploads,
        )
        returnValue(service) class _TorSocksFactory(Factory):
    protocol = _TorSocksProtocol

    # XXX should do validation on this stuff so we get errors before
    # building the protocol
    def __init__(self, *args, **kw):
        self._args = args
        self._kw = kw
        self._host = None
        self._when_connected = util.SingleObserver()

    def _get_address(self):
        """
        Returns a Deferred that fires with the transport's getHost()
        when this SOCKS protocol becomes connected.
        """
        return self._when_connected.when_fired()

    def _did_connect(self, host):
        self._host = host
        self._when_connected.fire(host)

    def buildProtocol(self, addr):
        p = self.protocol(*self._args, **self._kw)
        p.factory = self
        return p class OnionServiceTest(unittest.TestCase):
        class Bad(Exception):
class EphemeralHiddenServiceTest(unittest.TestCase):
class AuthenticatedFilesystemHiddenServiceTest(unittest.TestCase):
    def test_prop224_private_key(self):
    def test_set_ports(self):
    def test_set_dir(self):
    def test_dir_ioerror(self):
    def test_dir_ioerror_v3(self):
    def test_unknown_version(self):
    def test_ephemeral_given_key(self):
    def test_ephemeral_key_whitespace0(self):
    def test_ephemeral_key_whitespace1(self):
    def test_ephemeral_v3_no_key(self):
    def test_ephemeral_v3_ip_addr_tuple(self):
    def test_ephemeral_v3_non_anonymous(self):
    def test_ephemeral_v3_ip_addr_tuple_non_local(self):
    def test_ephemeral_v3_wrong_key_type(self):
    def test_ephemeral_ports_not_a_list(self):
    def test_ephemeral_ports_not_strings(self):
    def test_ephemeral_ports_no_spaces(self):
    def test_ephemeral_ports_no_colon(self):
    def test_ephemeral_ports_non_local(self):
    def test_ephemeral_ports_not_an_int(self):
    def test_filesystem_wrong_ports(self):
    def test_descriptor_all_uploads_fail(self):
        def progress(*args):
    def test_ephemeral_bad_return_value(self):
        def progress(*args):
        def check(f):
    def test_ephemeral_remove(self):
    def test_ephemeral_remove_not_ok(self):
    def test_ephemeral_ver_option(self):
    def test_ephemeral_extra_kwargs(self):
    def test_ephemeral_auth_stealth(self):
    def test_old_tor_version(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress_await_all(self):
        def my_progress(a, b, c):
    def test_ephemeral_auth_basic(self):
    def test_ephemeral_auth_basic_remove_fails(self):
    def test_ephemeral_auth_basic_bad_name(self):
    def test_ephemeral_auth_unknown(self):
    def test_ephemeral_ports_bad0(self):
    def test_ephemeral_ports_bad1(self):
    def test_ephemeral_ports_bad2(self):
    def test_defaults(self):
    def test_wrong_blob(self):
    def test_add(self):
    def test_add_keyblob(self):
    def test_descriptor_wait(self):
    def test_remove(self):
    def test_remove_error(self):
    def test_failed_upload(self):
    def test_single_failed_upload(self):
    def setUp(self):
    def test_create_progress_old_tor(self):
        def progress(pct, tag, msg):
    def test_unknown_auth_type(self):
    def test_bad_client_name(self):
    def test_get_client_missing(self):
    def test_get_client(self):
    def test_get_client_private_key_error(self):
    def test_get_client_expected_not_found(self): class OnionServiceTest(unittest.TestCase):
        class Bad(Exception):
class EphemeralHiddenServiceTest(unittest.TestCase):
class AuthenticatedFilesystemHiddenServiceTest(unittest.TestCase):
    def test_prop224_private_key(self):
    def test_set_ports(self):
    def test_set_dir(self):
    def test_dir_ioerror(self):
    def test_dir_ioerror_v3(self):
    def test_unknown_version(self):
    def test_ephemeral_given_key(self):
    def test_ephemeral_key_whitespace0(self):
    def test_ephemeral_key_whitespace1(self):
    def test_ephemeral_v3_no_key(self):
    def test_ephemeral_v3_ip_addr_tuple(self):
    def test_ephemeral_v3_non_anonymous(self):
    def test_ephemeral_v3_ip_addr_tuple_non_local(self):
    def test_ephemeral_v3_wrong_key_type(self):
    def test_ephemeral_ports_not_a_list(self):
    def test_ephemeral_ports_not_strings(self):
    def test_ephemeral_ports_no_spaces(self):
    def test_ephemeral_ports_no_colon(self):
    def test_ephemeral_ports_non_local(self):
    def test_ephemeral_ports_not_an_int(self):
    def test_filesystem_wrong_ports(self):
    def test_descriptor_all_uploads_fail(self):
        def progress(*args):
    def test_ephemeral_bad_return_value(self):
        def progress(*args):
        def check(f):
    def test_ephemeral_remove(self):
    def test_ephemeral_remove_not_ok(self):
    def test_ephemeral_ver_option(self):
    def test_ephemeral_extra_kwargs(self):
    def test_ephemeral_auth_stealth(self):
    def test_old_tor_version(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress_await_all(self):
        def my_progress(a, b, c):
    def test_ephemeral_auth_basic(self):
    def test_ephemeral_auth_basic_remove_fails(self):
    def test_ephemeral_auth_basic_bad_name(self):
    def test_ephemeral_auth_unknown(self):
    def test_ephemeral_ports_bad0(self):
    def test_ephemeral_ports_bad1(self):
    def test_ephemeral_ports_bad2(self):
    def test_defaults(self):
    def test_wrong_blob(self):
    def test_add(self):
    def test_add_keyblob(self):
    def test_descriptor_wait(self):
    def test_remove(self):
    def test_remove_error(self):
    def test_failed_upload(self):
    def test_single_failed_upload(self):
    def setUp(self):
    def test_create_progress_old_tor(self):
        def progress(pct, tag, msg):
    def test_unknown_auth_type(self):
    def test_bad_client_name(self):
    def test_get_client_missing(self):
    def test_get_client(self):
    def test_get_client_private_key_error(self):
    def test_get_client_expected_not_found(self): class OnionServiceTest(unittest.TestCase):
        class Bad(Exception):
class EphemeralHiddenServiceTest(unittest.TestCase):
class AuthenticatedFilesystemHiddenServiceTest(unittest.TestCase):
    def test_prop224_private_key(self):
    def test_set_ports(self):
    def test_set_dir(self):
    def test_dir_ioerror(self):
    def test_dir_ioerror_v3(self):
    def test_unknown_version(self):
    def test_ephemeral_given_key(self):
    def test_ephemeral_key_whitespace0(self):
    def test_ephemeral_key_whitespace1(self):
    def test_ephemeral_v3_no_key(self):
    def test_ephemeral_v3_ip_addr_tuple(self):
    def test_ephemeral_v3_non_anonymous(self):
    def test_ephemeral_v3_ip_addr_tuple_non_local(self):
    def test_ephemeral_v3_wrong_key_type(self):
    def test_ephemeral_ports_not_a_list(self):
    def test_ephemeral_ports_not_strings(self):
    def test_ephemeral_ports_no_spaces(self):
    def test_ephemeral_ports_no_colon(self):
    def test_ephemeral_ports_non_local(self):
    def test_ephemeral_ports_not_an_int(self):
    def test_filesystem_wrong_ports(self):
    def test_descriptor_all_uploads_fail(self):
        def progress(*args):
    def test_ephemeral_bad_return_value(self):
        def progress(*args):
        def check(f):
    def test_ephemeral_remove(self):
    def test_ephemeral_remove_not_ok(self):
    def test_ephemeral_ver_option(self):
    def test_ephemeral_extra_kwargs(self):
    def test_ephemeral_auth_stealth(self):
    def test_old_tor_version(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress(self):
        def my_progress(a, b, c):
    def test_tor_version_v3_progress_await_all(self):
        def my_progress(a, b, c):
    def test_ephemeral_auth_basic(self):
    def test_ephemeral_auth_basic_remove_fails(self):
    def test_ephemeral_auth_basic_bad_name(self):
    def test_ephemeral_auth_unknown(self):
    def test_ephemeral_ports_bad0(self):
    def test_ephemeral_ports_bad1(self):
    def test_ephemeral_ports_bad2(self):
    def test_defaults(self):
    def test_wrong_blob(self):
    def test_add(self):
    def test_add_keyblob(self):
    def test_descriptor_wait(self):
    def test_remove(self):
    def test_remove_error(self):
    def test_failed_upload(self):
    def test_single_failed_upload(self):
    def setUp(self):
    def test_create_progress_old_tor(self):
        def progress(pct, tag, msg):
    def test_unknown_auth_type(self):
    def test_bad_client_name(self):
    def test_get_client_missing(self):
    def test_get_client(self):
    def test_get_client_private_key_error(self):
    def test_get_client_expected_not_found(self): class FakeControlProtocol(object):
    """
    This is a little weird, but in most tests the answer at the top of
    the list is sent back immediately in an already-called
    Deferred. However, if the answer list is empty at the time of the
    call, instead the returned Deferred is added to the pending list
    and answer_pending() may be called to have the next Deferred
    fire. (see test_accept_all_postbootstrap for an example).

    It is done this way in case we need to have some other code run
    between the get_conf (or whatever) and the callback -- if the
    Deferred is already-fired when get_conf runs, there's a Very Good
    Chance (always?) that the callback just runs right away.
    """

    def __init__(self, answers):
        self.answers = answers
        self.pending = []
        self.post_bootstrap = defer.succeed(self)
        self.on_disconnect = defer.Deferred()
        self.sets = []
        self.events = {}  #: event type -> callback
        self.pending_events = {}  #: event type -> list
        self.is_owned = -1
        self.version = "0.2.8.0"
        # XXX can we get rud of 'commands' and just use pending?
        self.commands = []

    def queue_command(self, cmd):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            self.commands.append((cmd, d))
            return d

        d = defer.succeed(self.answers[0])
        self.commands.append((cmd, d))
        self.answers = self.answers[1:]
        return d

    def event_happened(self, event_type, *args):
        '''
        Use this in your tests to send 650 events when an event-listener
        is added.  XXX Also if we've *already* added one? Do that if
        there's a use-case for it
        '''
        if event_type in self.events:
            self.events[event_type](*args)
        elif event_type in self.pending_events:
            self.pending_events[event_type].append(args)
        else:
            self.pending_events[event_type] = [args]

    def answer_pending(self, answer):
        d = self.pending[0]
        self.pending = self.pending[1:]
        d.callback(answer)

    def get_info(self, info):
        a = self.answers.pop()
        return defer.succeed(a)

    def get_info_raw(self, info):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            return d

        d = defer.succeed(self.answers[0])
        self.answers = self.answers[1:]
        return d

    @defer.inlineCallbacks
    def get_info_incremental(self, info, cb):
        text = yield self.get_info_raw(info)
        for line in text.split('\r\n'):
            cb(line)
        defer.returnValue('')  # FIXME uh....what's up at torstate.py:350?

    def get_conf(self, info):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            return d

        d = defer.succeed(self.answers[0])
        self.answers = self.answers[1:]
        return d

    get_conf_raw = get_conf  # up to test author ensure the answer is a raw string
    get_conf_single = get_conf

    def set_conf(self, *args):
        for i in range(0, len(args), 2):
            self.sets.append((args[i], args[i + 1]))
        return defer.succeed('')

    def add_event_listener(self, nm, cb):
        self.events[nm] = cb
        if nm in self.pending_events:
            for event in self.pending_events[nm]:
                cb(*event)
        return defer.succeed(None)

    def remove_event_listener(self, nm, cb):
        del self.events[nm]
import os
import sys
from mock import patch
from mock import Mock, MagicMock
from unittest import skipIf
from binascii import b2a_base64
from zope.interface import implementer, directlyProvides
from twisted.trial import unittest
from twisted.test import proto_helpers
from twisted.internet import defer, error, tcp, unix
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.internet.endpoints import UNIXClientEndpoint
from twisted.internet.endpoints import serverFromString
from twisted.internet.endpoints import clientFromString
from twisted.python.failure import Failure
from twisted.internet.error import ConnectionRefusedError
from twisted.internet.interfaces import IStreamClientEndpoint
from twisted.internet.interfaces import IReactorCore
from twisted.internet.interfaces import IProtocol
from twisted.internet.interfaces import IReactorTCP
from twisted.internet.interfaces import IListeningPort
from twisted.internet.interfaces import IAddress
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from txtorcon import TorControlProtocol
from txtorcon import TorConfig
from txtorcon import TCPHiddenServiceEndpoint
from txtorcon import TorClientEndpoint
from txtorcon import IProgressProvider
from txtorcon import TorOnionAddress
from txtorcon.onion import IAuthenticatedOnionClients
from txtorcon.onion import IOnionService
from txtorcon.onion import _compute_permanent_id
from txtorcon import AuthStealth
from txtorcon import AuthBasic
from txtorcon.util import NoOpProtocolFactory
from txtorcon.util import SingleObserver
from txtorcon.endpoints import get_global_tor                       # FIXME
from txtorcon.endpoints import _create_socks_endpoint
from txtorcon.circuit import TorCircuitEndpoint, _get_circuit_attacher
from txtorcon.controller import Tor
from txtorcon.socks import _TorSocksFactory
from . import util
from .test_onion import _test_private_key       # put in testutil?
from .test_onion import _test_private_key_blob  # put in testutil?
from .test_onion import _test_onion_id          # put in testutil?
from txtorcon.testutil import FakeControlProtocol
        from txtorcon import endpoints
        from txtorcon import endpoints
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        from txtorcon import torconfig
        get_global_tor(
            self.reactor,
            _tor_launcher=lambda react, config, progress_updates=None: defer.succeed(config)
        )
        ep = serverFromString(
            self.reactor,
            'onion:88:localPort=1234:hiddenServiceDir=~/blam/blarg'
        )
        # would be nice to have a fixed path here, but then would have
        # to run as a known user :/
        # maybe using the docker stuff to run integration tests better here?
        self.assertEqual(
            os.path.expanduser('~/blam/blarg'),
            ep.hidden_service_dir
        )

    def test_parse_relative_path(self, ftb):
        # this makes sure we convert a relative path to absolute
        # hiddenServiceDir args. see Issue #77

        # make sure we have a valid thing from get_global_tor without
        # actually launching tor
        config = TorConfig()
        config.post_bootstrap = defer.succeed(config)
        torconfig._global_tor_config = None
        get_global_tor(
            self.reactor,
            _tor_launcher=lambda react, config, progress_updates=None: defer.succeed(config)
        )

        orig = os.path.realpath('.')
        try:
            with util.TempDir() as t:
                t = str(t)
                os.chdir(t)
                os.mkdir(os.path.join(t, 'foo'))
                hsdir = os.path.join(t, 'foo', 'blam')
                os.mkdir(hsdir)

                ep = serverFromString(
                    self.reactor,
                    'onion:88:localPort=1234:hiddenServiceDir=foo/blam'
                )
                self.assertEqual(
                    os.path.realpath(hsdir),
                    ep.hidden_service_dir
                )

        finally:
            os.chdir(orig)

    def test_illegal_arg_ephemeral_auth(self, ftb):
        # XXX I think Tor does actually support this now?
        reactor = Mock()
        config = Mock()
        with self.assertRaises(ValueError) as ctx:
            TCPHiddenServiceEndpoint(
                reactor, config, 80,
                stealth_auth=['alice', 'bob'],
                ephemeral=True,
            )
        self.assertIn(
            "onion services don't support 'stealth' auth",
            str(ctx.exception),
        )

    def test_illegal_arg_ephemeral_hsdir(self, ftb):
        reactor = Mock()
        config = Mock()
        with self.assertRaises(ValueError) as ctx:
            TCPHiddenServiceEndpoint(
                reactor, config, 80,
                ephemeral=True,
                hidden_service_dir='/tmp/foo',
            )
        self.assertIn(
            "Specifying 'hidden_service_dir' is incompatible",
            str(ctx.exception),
        )

    def test_illegal_arg_disk_privkey(self, ftb):
        reactor = Mock()
        config = Mock()
        with self.assertRaises(ValueError) as ctx:
            TCPHiddenServiceEndpoint(
                reactor, config, 80,
                ephemeral=False,
                private_key=b'something',
            )
        self.assertIn(
            "only understood for ephemeral services",
            str(ctx.exception),
        )

    @defer.inlineCallbacks
    def test_illegal_arg_torconfig(self, ftb):

        class Foo(object):
            pass
        config = Foo()

        ep = TCPHiddenServiceEndpoint(self.reactor, config, 123)
        factory = Mock()
        with self.assertRaises(ValueError) as ctx:
            yield ep.listen(factory)

        self.assertIn(
            "Expected a TorConfig instance but",
            str(ctx.exception)
        )

    @skipIf('pypy' in sys.version.lower(), "Weird OpenSSL+PyPy problem on Travis")
    @defer.inlineCallbacks
    def test_basic_auth_ephemeral(self, ftb):
        '''
        '''
        ep = TCPHiddenServiceEndpoint(
            self.reactor, self.config, 123,
            ephemeral=True,
            auth=AuthBasic(['alice', 'bob']),

            private_key=_test_private_key_blob,