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 TorProtocolError(RuntimeError):
    """
    Happens on 500-level responses in the protocol, almost certainly
    in an errback chain.

    :ivar code: the actual error code
    :ivar text: other text from the protocol
    """

    def __init__(self, code, text):
        self.code = code
        self.text = text
        super(TorProtocolError, self).__init__(text)

    def __str__(self):
        return str(self.code) + ' ' + self.text class TorState(object):
    """
    This tracks the current state of Tor using a TorControlProtocol.

    On setup it first queries the initial state of streams and
    circuits. It then asks for updates via the listeners. It requires
    an ITorControlProtocol instance. The control protocol doesn't need
    to be bootstrapped yet. The Deferred .post_boostrap is driggered
    when the TorState instance is fully ready to go.  The easiest way
    is to use the helper method
    :func:`txtorcon.build_tor_connection`. For details, see the
    implementation of that.

    You may add an :class:`txtorcon.interface.IStreamAttacher` to
    provide a custom mapping for Strams to Circuits (by default Tor
    picks by itself).

    This is also a good example of the various listeners, and acts as
    an :class:`txtorcon.interface.ICircuitContainer` and
    :class:`txtorcon.interface.IRouterContainer`.

    :cvar DO_NOT_ATTACH:
        Constant to return from an IAttacher indicating you don't want to
        attach this stream at all.

    """

    @classmethod
    def from_protocol(cls, protocol, **kw):
        '''
        Create a new, boot-strapped TorState from a TorControlProtocol
        instance.

        :return: a Deferred that fires with a TorState instance
        '''
        state = TorState(protocol, bootstrap=True)
        return state.post_bootstrap

    def __init__(self, protocol, bootstrap=True):
        self.protocol = ITorControlProtocol(protocol)
        # fixme could use protocol.on_disconnect to re-connect; see issue #3

        # could override these to get your own Circuit/Stream subclasses
        # to track these things
        self.circuit_factory = Circuit
        self.stream_factory = Stream

        self._attacher = None
        """If set, provides
        :class:`txtorcon.interface.IStreamAttacher` to attach new
        streams we hear about."""

        self.tor_binary = 'tor'

        self.circuit_listeners = []
        self.stream_listeners = []

        self.addrmap = AddrMap()
        #: keys on id (integer)
        self.circuits = {}

        #: keys on id (integer)
        self.streams = {}

        #: list of unique routers
        self.all_routers = set()

        #: keys by hexid (string) and by unique names
        self.routers = {}
        self._old_routers = {}

        #: keys on name, value always list (many duplicate "Unnamed"
        #: routers, for example)
        self.routers_by_name = {}

        #: keys by hexid (string)
        self.routers_by_hash = {}

        #: potentially-usable as entry guards, I think? (any router
        #: with 'Guard' flag)
        self.guards = {}

        #: from GETINFO entry-guards, our current entry guards
        self.entry_guards = {}

        #: list of entry guards we didn't parse out
        self.unusable_entry_guards = []

        #: keys by name
        self.authorities = {}

        #: see set_attacher
        self._cleanup = None

        self._network_status_parser = MicrodescriptorParser(self._create_router)

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

    def _create_router(self, **kw):
        id_hex = hexIdFromHash(kw['idhash'])
        try:
            router = self._old_routers[id_hex]
        except KeyError:
            router = Router(self.protocol)

        self.routers[id_hex] = router
        router.from_consensus = True
        router.update(
            kw['nickname'],
            kw['idhash'],
            kw['orhash'],
            kw['modified'],
            kw['ip'],
            kw['orport'],
            kw['dirport'],
        )
        router.flags = kw.get('flags', [])
        if 'bandwidth' in kw:
            router.bandwidth = kw['bandwidth']
        if 'ip_v6' in kw:
            router.ip_v6.extend(kw['ip_v6'])

        if 'guard' in router.flags:
            self.guards[router.id_hex] = router
        if 'authority' in router.flags:
            self.authorities[router.name] = router

        if router.name in self.routers:
            self.routers[router.name] = None

        else:
            self.routers[router.name] = router

        if router.name in self.routers_by_name:
            self.routers_by_name[router.name].append(router)

        else:
            self.routers_by_name[router.name] = [router]

        self.routers[router.id_hex] = router
        self.routers_by_hash[router.id_hex] = router
        self.all_routers.add(router)

    @defer.inlineCallbacks
    def _bootstrap(self, arg=None):
        "This takes an arg so we can use it as a callback (see __init__)."

        # update list of routers (must be before we do the
        # circuit-status)

        # look out! we're depending on get_info_incremental returning
        # *lines*, which isn't documented -- but will be true because
        # TorControlProtocol is a LineReceiver...
        yield self.protocol.get_info_incremental(
            'ns/all',
            self._network_status_parser.feed_line,
        )
        self._network_status_parser.done()

        # update list of existing circuits
        cs = yield self.protocol.get_info_raw('circuit-status')
        self._circuit_status(cs)

        # update list of streams
        ss = yield self.protocol.get_info_raw('stream-status')
        self._stream_status(ss)

        # update list of existing address-maps
        key = 'address-mappings/all'
        am = yield self.protocol.get_info_raw(key)
        # strip addressmappsings/all= and OK\n from raw data
        am = am[len(key) + 1:]
        for line in am.split('\n'):
            if len(line.strip()) == 0:
                continue            # FIXME
            self.addrmap.update(line)

        yield self._add_events()

        entries = yield self.protocol.get_info_raw("entry-guards")
        for line in entries.split('\n')[1:]:
            if len(line.strip()) == 0 or line.strip() == 'OK':
                # XXX does this ever really happen?
                continue
            args = line.split()
            (name, status) = args[:2]
            name = name[:41]

            # this is sometimes redundant, as a missing entry guard
            # usually means it won't be in our list of routers right
            # now, but just being on the safe side
            if status.lower() != 'up':
                self.unusable_entry_guards.append(line)
                continue

            try:
                self.entry_guards[name] = self.router_from_id(name)
            except KeyError:
                self.unusable_entry_guards.append(line)

        # in case process/pid doesn't exist and we don't know the PID
        # because we own it, we just leave it as 0 (previously
        # guessed using psutil, but that only works if there's
        # exactly one tor running anyway)
        try:
            pid = yield self.protocol.get_info_raw("process/pid")
        except TorProtocolError:
            pid = None
        self.tor_pid = 0
        if pid:
            try:
                pid = parse_keywords(pid)['process/pid']
                self.tor_pid = int(pid)
            except Exception:  # fixme: ValueError and KeyError ..?
                self.tor_pid = 0
        if not self.tor_pid and self.protocol.is_owned:
            self.tor_pid = self.protocol.is_owned

        self.post_bootstrap.callback(self)
        self.post_boostrap = None

    # XXX this should be hidden as _undo_attacher
    def undo_attacher(self):
        """
        Shouldn't Tor handle this by turning this back to 0 if the
        controller that twiddled it disconnects?
        """

        return self.protocol.set_conf("__LeaveStreamsUnattached", 0)

    def set_attacher(self, attacher, myreactor):
        """
        Provide an :class:`txtorcon.interface.IStreamAttacher` to
        associate streams to circuits.

        You are Strongly Encouraged to **not** use this API directly,
        and instead use :meth:`txtorcon.Circuit.stream_via` or
        :meth:`txtorcon.Circuit.web_agent` instead. If you do need to
        use this API, it's an error if you call either of the other
        two methods.

        This won't get turned on until after bootstrapping is
        completed. ('__LeaveStreamsUnattached' needs to be set to '1'
        and the existing circuits list needs to be populated).
        """

        react = IReactorCore(myreactor)
        if attacher:
            if self._attacher is attacher:
                return
            if self._attacher is not None:
                raise RuntimeError(
                    "set_attacher called but we already have an attacher"
                )
            self._attacher = IStreamAttacher(attacher)
        else:
            self._attacher = None

        if self._attacher is None:
            d = self.undo_attacher()
            if self._cleanup:
                react.removeSystemEventTrigger(self._cleanup)
                self._cleanup = None

        else:
            d = self.protocol.set_conf("__LeaveStreamsUnattached", "1")
            self._cleanup = react.addSystemEventTrigger(
                'before', 'shutdown',
                self.undo_attacher,
            )
        return d

    # noqa
    stream_close_reasons = {
        'REASON_MISC': 1,               # (catch-all for unlisted reasons)
        'REASON_RESOLVEFAILED': 2,      # (couldn't look up hostname)
        'REASON_CONNECTREFUSED': 3,     # (remote host refused connection) [*]
        'REASON_EXITPOLICY': 4,         # (OR refuses to connect to host or port)
        'REASON_DESTROY': 5,            # (Circuit is being destroyed)
        'REASON_DONE': 6,               # (Anonymized TCP connection was closed)
        'REASON_TIMEOUT': 7,            # (Connection timed out, or OR timed out while connecting)
        'REASON_NOROUTE': 8,            # (Routing error while attempting to contact destination)
        'REASON_HIBERNATING': 9,        # (OR is temporarily hibernating)
        'REASON_INTERNAL': 10,          # (Internal error at the OR)
        'REASON_RESOURCELIMIT': 11,     # (OR has no resources to fulfill request)
        'REASON_CONNRESET': 12,         # (Connection was unexpectedly reset)
        'REASON_TORPROTOCOL': 13,       # (Sent when closing connection because of Tor protocol violations.)
        'REASON_NOTDIRECTORY': 14}      # (Client sent RELAY_BEGIN_DIR to a non-directory relay.)

    def close_stream(self, stream, reason='REASON_MISC', **kwargs):
        """
        This sends a STREAMCLOSE command, using the specified reason
        (either an int or one of the 14 strings in section 6.3 of
        tor-spec.txt if the argument is a string). Any kwards are
        passed through as flags if they evaluated to true
        (e.g. "SomeFlag=True"). Currently there are none that Tor accepts.
        """

        if type(stream) != int:
            # assume it's a Stream instance
            stream = stream.id
        try:
            reason = int(reason)
        except ValueError:
            try:
                reason = TorState.stream_close_reasons[reason]
            except KeyError:
                raise ValueError(
                    'Unknown stream close reason "%s"' % str(reason)
                )

        flags = flags_from_dict(kwargs)

        # stream is now an ID no matter what we passed in
        cmd = 'CLOSESTREAM %d %d%s' % (stream, reason, flags)
        return self.protocol.queue_command(cmd)

    def close_circuit(self, circid, **kwargs):
        """
        This sends a CLOSECIRCUIT command, using any keyword arguments
        passed as the Flags (currently, that is just 'IfUnused' which
        means to only close the circuit when it is no longer used by
        any streams).

        :param circid:
            Either a circuit-id (int) or a Circuit instance

        :return:
            a Deferred which callbacks with the result of queuing the
            command to Tor (usually "OK"). If you want to instead know
            when the circuit is actually-gone, see
            :meth:`Circuit.close <txtorcon.circuit.Circuit.close>`
        """

        if type(circid) != int:
            # assume it's a Circuit instance
            circid = circid.id
        flags = flags_from_dict(kwargs)
        return self.protocol.queue_command(
            'CLOSECIRCUIT %s%s' % (circid, flags)
        )

    def on_circuit_new(self, callback):
        """
        :param callback: will be called (with 'circuit' instance) when a
            CIRC NEW event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_new = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def on_circuit_launched(self, callback):
        """
        :param callback: will be called (with 'circuit' instance) when a
             CIRC LAUNCHED event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_launched = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def on_circuit_extend(self, callback):
        """
        :param callback: will be called (with 'circuit' and 'router'
            instances) when a CIRC EXTENDED event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_extend = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def on_circuit_built(self, callback):
        """
        :param callback: will be called (with 'circuit' instance) when a
            CIRC BUILT event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_built = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def on_circuit_closed(self, callback):
        """
        :param callback: will be called (with 'circuit' instance, and
            arbitrary kwargs) when a CIRC CLOSED event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_closed = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def on_circuit_failed(self, callback):
        """
        :param callback: will be called (with 'circuit' instance, and
            arbitrary kwargs) when a CIRC FAILED event happens
        """
        listener = CircuitListenerMixin()
        listener.circuit_failed = callback
        self.add_circuit_listener(listener)
        return callback  # so we can be used as a listener

    def add_circuit_listener(self, icircuitlistener):
        """
        Adds a new instance of :class:`txtorcon.interface.ICircuitListener` which
        will receive updates for all existing and new circuits.
        """
        listen = ICircuitListener(icircuitlistener)
        for circ in self.circuits.values():
            circ.listen(listen)
        self.circuit_listeners.append(listen)

    def on_stream_new(self, callback):
        """
        :param callback: will be called (with 'stream' instance) when a
            STREAM NEW event happens.
        """
        listener = StreamListenerMixin()
        listener.stream_new = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def on_stream_succeeded(self, callback):
        """
        :param callback: will be called (with 'stream' instance) when a
            STREAM SUCCEEDED event happens.
        """
        listener = StreamListenerMixin()
        listener.stream_succeeded = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def on_stream_attach(self, callback):
        """
        :param callback: will be called (with 'stream' and 'circuit'
            instances) when a stream is attached to a circuit
        """
        listener = StreamListenerMixin()
        listener.stream_attach = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def on_stream_detach(self, callback):
        """
        :param callback: will be called (with 'stream' instance and
            arbitrary kwargs) when a STREAM DETACHED happens
        """
        listener = StreamListenerMixin()
        listener.stream_detach = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def on_stream_closed(self, callback):
        """
        :param callback: will be called (with 'stream' instance and
            arbitrary kwargs) when a STREAM CLOSED event happens.
        """
        listener = StreamListenerMixin()
        listener.stream_closed = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def on_stream_failed(self, callback):
        """
        :param callback: will be called (with 'stream' instance and
            arbitrary kwargs) when a STREAM FAILED event happens.
        """
        listener = StreamListenerMixin()
        listener.stream_failed = callback
        self.add_stream_listener(listener)
        return callback  # so we can be used as a listener

    def add_stream_listener(self, istreamlistener):
        """
        Adds a new instance of :class:`txtorcon.interface.IStreamListener` which
        will receive updates for all existing and new streams.
        """
        listen = IStreamListener(istreamlistener)
        for stream in self.streams.values():
            stream.listen(listen)
        self.stream_listeners.append(listen)

    def _find_circuit_after_extend(self, x):
        ex, circ_id = x.split()
        if ex != 'EXTENDED':
            raise RuntimeError('Expected EXTENDED, got "%s"' % x)
        circ_id = int(circ_id)
        circ = self._maybe_create_circuit(circ_id)
        circ.update([str(circ_id), 'EXTENDED'])
        return circ

    def build_circuit(self, routers=None, using_guards=True, purpose=None):
        """
        Builds a circuit consisting of exactly the routers specified,
        in order.  This issues an EXTENDCIRCUIT call to Tor with all
        the routers specified.

        :param routers: a list of Router instances which is the path
            desired. To allow Tor to choose the routers itself, pass
            None (the default) for routers.

        :param using_guards: A warning is issued if the first router
            isn't in self.entry_guards.

        :return:
            A Deferred that will callback with a Circuit instance
            (with the .id member being valid, and probably nothing
            else).
        """

        if routers is None or routers == []:
            cmd = "EXTENDCIRCUIT 0"

        else:
            if using_guards and routers[0] not in self.entry_guards.values():
                warnings.warn(
                    "Circuit doesn't start with a guard: %s" % routers,
                    RuntimeWarning
                )
            cmd = "EXTENDCIRCUIT 0 "
            first = True
            for router in routers:
                if first:
                    first = False
                else:
                    cmd += ','
                # XXX should we really accept bytes here?
                if isinstance(router, bytes) and len(router) == 40 \
                   and hashFromHexId(router):
                    cmd += router.decode('utf8')
                else:
                    cmd += router.id_hex[1:]

            if purpose is not None:
                cmd += " purpose={}".format(purpose)
        d = self.protocol.queue_command(cmd)
        d.addCallback(self._find_circuit_after_extend)
        return d

    DO_NOT_ATTACH = object()

    # @defer.inlineCallbacks  (this method is async, be nice to mark it ...)
    def _maybe_attach(self, stream):
        """
        If we've got a custom stream-attachment instance (see
        set_attacher) this will ask it for the appropriate
        circuit. Note that we ignore .exit URIs and let Tor deal with
        those (by passing circuit ID 0).

        The stream attacher is allowed to return a Deferred which will
        callback with the desired circuit.

        You may return the special object DO_NOT_ATTACH which will
        cause the circuit attacher to simply ignore the stream
        (neither attaching it, nor telling Tor to attach it).
        """

        if self._attacher is None:
            return None

        if stream.target_host is not None \
           and '.exit' in stream.target_host:
            # we want to totally ignore .exit URIs as these are
            # used to specify a particular exit node, and trying
            # to do STREAMATTACH on them will fail with an error
            # from Tor anyway.
            txtorlog.msg("ignore attacher:", stream)
            return

        # handle async or sync .attach() the same
        circ_d = defer.maybeDeferred(
            self._attacher.attach_stream,
            stream, self.circuits,
        )
        circ_d.addCallback(maybe_coroutine)

        # actually do the attachment logic; .attach() can return 3 things:
        #    1. None: let Tor do whatever it wants
        #    2. DO_NOT_ATTACH: don't attach the stream at all
        #    3. Circuit instance: attach to the provided circuit
        def issue_stream_attach(circ):
            txtorlog.msg("circuit:", circ)
            if circ is None or circ is TorState.DO_NOT_ATTACH:
                # tell Tor to do what it likes
                return self.protocol.queue_command(
                    u"ATTACHSTREAM {} 0".format(stream.id).encode("ascii")
                )

            else:
                # should get a Circuit instance; check it for suitability
                if not isinstance(circ, Circuit):
                    raise RuntimeError(
                        "IStreamAttacher.attach() must return a Circuit instance "
                        "(or None or DO_NOT_ATTACH): %s"
                    )
                if circ.id not in self.circuits:
                    raise RuntimeError(
                        "Attacher returned a circuit unknown to me."
                    )
                if circ.state != 'BUILT':
                    raise RuntimeError(
                        "Can only attach to BUILT circuits; %d is in %s." %
                        (circ.id, circ.state)
                    )
                # we've got a valid Circuit instance; issue the command
                return self.protocol.queue_command(
                    u"ATTACHSTREAM {} {}".format(stream.id, circ.id).encode("ascii")
                )

        circ_d.addCallback(issue_stream_attach)
        circ_d.addErrback(self._attacher_error)
        return circ_d

    def _attacher_error(self, fail):
        """
        not ideal, but there's not really a good way to let the caller
        handler errors :/ since we ultimately call this due to an
        async request from Tor. Mostly these errors will be logic or
        syntax errors in the caller's code anyway.

        tests monkey-patch this to reduce spew
        """
        print("Failure while attaching stream:", fail)
        return fail

    def _circuit_status(self, data):
        """Used internally as a callback for updating Circuit information"""

        data = data[len('circuit-status='):].split('\n')
        # sometimes there's a newline after circuit-status= and
        # sometimes not, so we get rid of it
        if len(data) and len(data[0].strip()) == 0:
            data = data[1:]

        for line in data:
            self._circuit_update(line)

    def _stream_status(self, data):
        "Used internally as a callback for updating Stream information"
        # there's a slight issue with a single-stream vs >= 2 streams,
        # in that in the latter case we have a line by itself with
        # "stream-status=" on it followed by the streams EXCEPT in the
        # single-stream case which has "stream-status=123 blahblah"
        # (i.e. the key + value on one line)

        lines = data.split('\n')
        if len(lines) == 1:
            d = lines[0][len('stream-status='):]
            # if there are actually 0 streams, then there's nothing
            # left to parse
            if len(d):
                self._stream_update(d)
        else:
            [self._stream_update(line) for line in lines[1:]]

    def _update_network_status(self, data):
        """
        Used internally as a callback for updating Router information
        from NEWCONSENSUS events.
        """

        # XXX why are we ever getting this with 0 data?
        if len(data):
            self._old_routers = self.routers
            self.routers = dict()
            self.all_routers = set()
            self.routers_by_hash = dict()
            self.routers_by_name = dict()
            for line in data.split('\n'):
                self._network_status_parser.feed_line(line)
            self._network_status_parser.done()

        txtorlog.msg(len(self.routers_by_name), "named routers found.")
        # remove any names we added that turned out to have dups
        remove_keys = set()
        for (k, v) in self.routers.items():
            if v is None:
                txtorlog.msg(len(self.routers_by_name[k]), "dups:", k)
                remove_keys.add(k)
        for k in remove_keys:
            del self.routers[k]

        txtorlog.msg(len(self.guards), "GUARDs")

    def _maybe_create_circuit(self, circ_id):
        if circ_id not in self.circuits:
            c = self.circuit_factory(self)
            c.listen(self)
            for listener in self.circuit_listeners:
                c.listen(listener)

        else:
            c = self.circuits[circ_id]
        return c

    def _circuit_update(self, line):
        """
        Used internally as a callback to update Circuit information
        from CIRC events.
        """

        # print("circuit_update", line)
        args = line.split()
        circ_id = int(args[0])

        c = self._maybe_create_circuit(circ_id)
        c.update(args)

    def _stream_update(self, line):
        """
        Used internally as a callback to update Stream information
        from STREAM events.
        """

        if line.strip() == 'stream-status=':
            # this happens if there are no active streams
            return

        args = line.split()
        assert len(args) >= 3

        stream_id = int(args[0])
        wasnew = False
        if stream_id not in self.streams:
            stream = self.stream_factory(self, self.addrmap)
            self.streams[stream_id] = stream
            stream.listen(self)
            for x in self.stream_listeners:
                stream.listen(x)
            wasnew = True
        self.streams[stream_id].update(args)

        # if the update closed the stream, it won't be in our list
        # anymore. FIXME: how can we ever hit such a case as the
        # first update being a CLOSE?
        if wasnew and stream_id in self.streams:
            self._maybe_attach(self.streams[stream_id])

    def _addr_map(self, addr):
        "Internal callback to update DNS cache. Listens to ADDRMAP."
        txtorlog.msg(" --> addr_map", addr)
        self.addrmap.update(addr)

    event_map = {
        'STREAM': '_stream_update',
        'CIRC': '_circuit_update',
        'NEWCONSENSUS': '_update_network_status',
        'ADDRMAP': '_addr_map',
    }

    @defer.inlineCallbacks
    def _add_events(self):
        """
        Add listeners for all the events the controller is interested in.
        """

        for (event, func_name) in self.event_map.items():
            yield self.protocol.add_event_listener(
                event,
                getattr(self, func_name),
            )

    # ICircuitContainer

    def find_circuit(self, circid):
        "ICircuitContainer API"
        return self.circuits[circid]

    # IRouterContainer

    def router_from_id(self, routerid):
        """IRouterContainer API"""

        try:
            return self.routers[routerid[:41]]

        except KeyError:
            if routerid[0] != '$':
                raise                   # just re-raise the KeyError

            router = Router(self.protocol)
            idhash = routerid[1:41]
            nick = ''
            is_named = False
            if len(routerid) > 41:
                nick = routerid[42:]
                is_named = routerid[41] == '='
            router.update(nick, hashFromHexId(idhash), '0' * 27, 'unknown',
                          'unknown', '0', '0')
            router.name_is_unique = is_named
            self.routers[router.id_hex] = router
            return router

    # implement IStreamListener

    def stream_new(self, stream):
        "IStreamListener: a new stream has been created"
        txtorlog.msg("stream_new", stream)

    def stream_succeeded(self, stream):
        "IStreamListener: stream has succeeded"
        txtorlog.msg("stream_succeeded", stream)

    def stream_attach(self, stream, circuit):
        """
        IStreamListener: the stream has been attached to a circuit. It
        seems you get an attach to None followed by an attach to real
        circuit fairly frequently. Perhaps related to __LeaveStreamsUnattached?
        """
        txtorlog.msg("stream_attach", stream.id,
                     stream.target_host, " -> ", circuit)

    def stream_detach(self, stream, **kw):
        """
        IStreamListener
        """
        txtorlog.msg("stream_detach", stream.id)

    def stream_closed(self, stream, **kw):
        """
        IStreamListener: stream has been closed (won't be in
        controller's list anymore)
        """

        txtorlog.msg("stream_closed", stream.id)
        del self.streams[stream.id]

    def stream_failed(self, stream, **kw):
        """
        IStreamListener: stream failed for some reason (won't be in
        controller's list anymore)
        """

        txtorlog.msg("stream_failed", stream.id)
        del self.streams[stream.id]

    # implement ICircuitListener

    def circuit_launched(self, circuit):
        "ICircuitListener API"
        txtorlog.msg("circuit_launched", circuit)
        self.circuits[circuit.id] = circuit

    def circuit_extend(self, circuit, router):
        "ICircuitListener API"
        txtorlog.msg("circuit_extend:", circuit.id, router)

    def circuit_built(self, circuit):
        "ICircuitListener API"
        txtorlog.msg(
            "circuit_built:", circuit.id,
            "->".join("%s.%s" % (x.name, x.location.countrycode) for x in circuit.path),
            circuit.streams
        )

    def circuit_new(self, circuit):
        "ICircuitListener API"
        txtorlog.msg("circuit_new:", circuit.id)
        self.circuits[circuit.id] = circuit

    def circuit_destroy(self, circuit):
        "Used by circuit_closed and circuit_failed (below)"
        txtorlog.msg("circuit_destroy:", circuit.id)
        circuit._when_built.fire(
            Failure(Exception("Destroying circuit; will never hit BUILT"))
        )
        del self.circuits[circuit.id]

    def circuit_closed(self, circuit, **kw):
        "ICircuitListener API"
        txtorlog.msg("circuit_closed", circuit)
        circuit._when_built.fire(
            Failure(
                CircuitBuildClosedError(_extract_reason(kw))
            )
        )
        self.circuit_destroy(circuit)

    def circuit_failed(self, circuit, **kw):
        "ICircuitListener API"
        txtorlog.msg("circuit_failed", circuit, str(kw))
        circuit._when_built.fire(
            Failure(
                CircuitBuildFailedError(_extract_reason(kw))
            )
        )
        self.circuit_destroy(circuit) class Stream(object):
    """
    Represents an active stream in Tor's state (:class:`txtorcon.TorState`).

    :ivar circuit:
        Streams will generally be attached to circuits pretty
        quickly. If they are attached, circuit will be a
        :class:`txtorcon.Circuit` instance or None if this stream
        isn't yet attached to a circuit.

    :ivar state:
        Tor's idea of the stream's state, one of:
          - NEW: New request to connect
          - NEWRESOLVE: New request to resolve an address
          - REMAP: Address re-mapped to another
          - SENTCONNECT: Sent a connect cell along a circuit
          - SENTRESOLVE: Sent a resolve cell along a circuit
          - SUCCEEDED: Received a reply; stream established
          - FAILED: Stream failed and not retriable
          - CLOSED: Stream closed
          - DETACHED: Detached from circuit; still retriable

    :ivar target_host:
        Something like www.example.com -- the host the stream is destined for.

    :ivar target_port:
        The port the stream will exit to.

    :ivar target_addr:
        Target address, looked up (usually) by Tor (e.g. 127.0.0.1).

    :ivar id:
        The ID of this stream, a number (or None if unset).
    """

    def __init__(self, circuitcontainer, addrmap=None):
        """
        :param circuitcontainer: an object which implements
            :class:`interface.ICircuitContainer`
        """

        self.circuit_container = ICircuitContainer(circuitcontainer)

        # FIXME: Sphinx doesn't seem to understand these variable
        # docstrings, so consolidate with above if Sphinx is the
        # answer -- actually it does, so long as the :ivar: things
        # are never mentioned it seems.

        self.id = None
        """An int, Tor's ID for this :class:`txtorcon.Circuit`"""

        self.state = None
        """A string, Tor's idea of the state of this
        :class:`txtorcon.Stream`"""

        self.target_host = None
        """Usually a hostname, but sometimes an IP address (e.g. when
        we query existing state from Tor)"""

        self.target_addr = None
        """If available, the IP address we're connecting to (if None,
        see target_host instead)."""

        self.target_port = 0
        """The port we're connecting to."""

        self.circuit = None
        """If we've attached to a :class:`txtorcon.Circuit`, this will
        be an instance of :class:`txtorcon.Circuit` (otherwise None)."""

        self.listeners = []
        """A list of all connected
        :class:`txtorcon.interface.IStreamListener` instances."""

        self.source_addr = None
        """If available, the address from which this Stream originated
        (e.g. local process, etc). See get_process() also."""

        self.source_port = 0
        """If available, the port from which this Stream
        originated. See get_process() also."""

        self.flags = {}
        """All flags from last update to this Stream. str->str"""

        self._closing_deferred = None
        """Internal. Holds Deferred that will callback when this
        stream is CLOSED, FAILED (or DETACHED??)"""

        self._addrmap = addrmap

    def listen(self, listen):
        """
        Attach an :class:`txtorcon.interface.IStreamListener` to this stream.

        See also :meth:`txtorcon.TorState.add_stream_listener` to
        listen to all streams.

        :param listen: something that knows
            :class:`txtorcon.interface.IStreamListener`
        """

        listener = IStreamListener(listen)
        if listener not in self.listeners:
            self.listeners.append(listener)

    def unlisten(self, listener):
        self.listeners.remove(listener)

    def close(self, **kw):
        """
        This asks Tor to close the underlying stream object. See
        :meth:`txtorcon.interface.ITorControlProtocol.close_stream`
        for details.

        Although Tor currently takes no flags, it allows you to; any
        keyword arguments are passed through as flags.

        NOTE that the callback delivered from this method only
        callbacks after the underlying stream is really destroyed
        (*not* just when the CLOSESTREAM command has successfully
        completed).
        """

        self._closing_deferred = defer.Deferred()

        def close_command_is_queued(*args):
            return self._closing_deferred
        d = self.circuit_container.close_stream(self, **kw)
        d.addCallback(close_command_is_queued)
        return self._closing_deferred

    def _create_flags(self, kw):
        """
        this clones the kw dict, adding a lower-case version of every key
        (duplicated in circuit.py; consider putting in util?)
        """

        flags = {}
        for k in kw.keys():
            flags[k] = kw[k]
            flags[k.lower()] = flags[k]
        return flags

    def update(self, args):
        if self.id is None:
            self.id = int(args[0])
        else:
            if self.id != int(args[0]):
                raise RuntimeError("Update for wrong stream.")

        kw = find_keywords(args)
        self.flags = kw

        if 'SOURCE_ADDR' in kw:
            last_colon = kw['SOURCE_ADDR'].rfind(':')
            self.source_addr = kw['SOURCE_ADDR'][:last_colon]
            if self.source_addr != '(Tor_internal)':
                self.source_addr = maybe_ip_addr(self.source_addr)
            self.source_port = int(kw['SOURCE_ADDR'][last_colon + 1:])

        self.state = args[1]
        # XXX why not using the state-machine stuff? ;)
        if self.state in ['NEW', 'NEWRESOLVE', 'SUCCEEDED']:
            if self.target_host is None:
                last_colon = args[3].rfind(':')
                self.target_host = args[3][:last_colon]
                self.target_port = int(args[3][last_colon + 1:])
                # target_host is often an IP address (newer tors? did
                # this change?) so we attempt to look it up in our
                # AddrMap and make it a name no matter what.
                if self._addrmap:
                    try:
                        h = self._addrmap.find(self.target_host)
                        self.target_host = h.name
                    except KeyError:
                        pass

            self.target_port = int(self.target_port)
            if self.state == 'NEW':
                if self.circuit is not None:
                    log.err(RuntimeError("Weird: circuit valid in NEW"))
                self._notify('stream_new', self)
            else:
                self._notify('stream_succeeded', self)

        elif self.state == 'REMAP':
            self.target_addr = maybe_ip_addr(args[3][:args[3].rfind(':')])

        elif self.state == 'CLOSED':
            if self.circuit:
                self.circuit.streams.remove(self)
            self.circuit = None
            self.maybe_call_closing_deferred()
            flags = self._create_flags(kw)
            self._notify('stream_closed', self, **flags)

        elif self.state == 'FAILED':
            if self.circuit:
                self.circuit.streams.remove(self)
            self.circuit = None
            self.maybe_call_closing_deferred()
            # build lower-case version of all flags
            flags = self._create_flags(kw)
            self._notify('stream_failed', self, **flags)

        elif self.state == 'SENTCONNECT':
            pass  # print 'SENTCONNECT',self,args

        elif self.state == 'DETACHED':
            if self.circuit:
                self.circuit.streams.remove(self)
                self.circuit = None

            # FIXME does this count as closed?
            # self.maybe_call_closing_deferred()
            flags = self._create_flags(kw)
            self._notify('stream_detach', self, **flags)

        elif self.state in ['NEWRESOLVE', 'SENTRESOLVE']:
            pass  # print self.state, self, args

        else:
            raise RuntimeError("Unknown state: %s" % self.state)

        # see if we attached to a circuit. I believe this only happens
        # on a SENTCONNECT or REMAP. DETACHED is excluded so we don't
        # immediately re-add the circuit we just detached from
        if self.state not in ['CLOSED', 'FAILED', 'DETACHED']:
            cid = int(args[2])
            if cid == 0:
                if self.circuit and self in self.circuit.streams:
                    self.circuit.streams.remove(self)
                self.circuit = None

            else:
                if self.circuit is None:
                    self.circuit = self.circuit_container.find_circuit(cid)
                    if self not in self.circuit.streams:
                        self.circuit.streams.append(self)
                        self._notify('stream_attach', self, self.circuit)

                else:
                    # XXX I am seeing this from torexitscan (*and*
                    # from my TorNS thing, so I think it's some kind
                    # of 'consistent' behavior out of Tor) so ... this
                    # is probably when we're doing stream-attachment
                    # stuff? maybe the stream gets assigned a circuit
                    # 'provisionally' and then it's changed?
                    # ...yup, looks like it!
                    if self.circuit.id != cid:
                        # you can get SENTCONNECT twice in a row (for example) with different circuit-ids if there is something (e.g. another txtorcon) doing R
                        log.err(
                            RuntimeError(
                                'Circuit ID changed from %d to %d state=%s.' %
                                (self.circuit.id, cid, self.state)
                            )
                        )

    def _notify(self, func, *args, **kw):
        """
        Internal helper. Calls the IStreamListener function 'func' with
        the given args, guarding around errors.
        """
        for x in self.listeners:
            try:
                getattr(x, func)(*args, **kw)
            except Exception:
                log.err()

    def maybe_call_closing_deferred(self):
        """
        Used internally to callback on the _closing_deferred if it
        exists.
        """

        if self._closing_deferred:
            self._closing_deferred.callback(self)
            self._closing_deferred = None

    def __str__(self):
        c = ''
        if self.circuit:
            c = 'on %d ' % self.circuit.id
        return "<Stream %s %d %s%s -> %s port %d>" % (self.state,
                                                      self.id,
                                                      c,
                                                      self.target_host,
                                                      str(self.target_addr),
                                                      self.target_port) class Circuit(object):
    """
    Used by :class:`txtorcon.TorState` to represent one of Tor's circuits.

    This is kept up-to-date by the :class`txtorcon.TorState` that owns it, and
    individual circuits can be listened to for updates (or listen to
    every one using :meth:`txtorcon.TorState.add_circuit_listener`)

    :ivar path:
        contains a list of :class:`txtorcon.Router` objects
        representing the path this Circuit takes. Mostly this will be
        3 or 4 routers long. Note that internally Tor uses single-hop
        paths for some things. See also the *purpose*
        instance-variable.

    :ivar streams:
        contains a list of Stream objects representing all streams
        currently attached to this circuit.

    :ivar state:
        contains a string from Tor describing the current state of the
        stream. From control-spec.txt section 4.1.1, these are:

            - LAUNCHED: circuit ID assigned to new circuit
            - BUILT: all hops finished, can now accept streams
            - EXTENDED: one more hop has been completed
            - FAILED: circuit closed (was not built)
            - CLOSED: circuit closed (was built)

    :ivar purpose:
        The reason this circuit was built. For most purposes, you'll
        want to look at `GENERAL` circuits only. Values can currently
        be one of (but see control-spec.txt 4.1.1):

            - GENERAL
            - HS_CLIENT_INTRO
            - HS_CLIENT_REND
            - HS_SERVICE_INTRO
            - HS_SERVICE_REND
            - TESTING
            - CONTROLLER

    :ivar id: The ID of this circuit, a number (or None if unset).

    """

    def __init__(self, routercontainer):
        """
        :param routercontainer:
            should implement :class:`txtorcon.interface.IRouterContainer`.
        """
        self.listeners = []
        self.router_container = IRouterContainer(routercontainer)
        self._torstate = routercontainer  # XXX FIXME
        self.path = []
        self.streams = []
        self.purpose = None
        self.id = None
        self.state = 'UNKNOWN'
        self.build_flags = []
        self.flags = {}

        # this is used to hold a Deferred that will callback() when
        # this circuit is being CLOSED or FAILED.
        self._closing_deferred = None
        # XXX ^ should probably be when_closed() etc etc...

        # caches parsed value for time_created()
        self._time_created = None

        # all notifications for when_built, when_closed
        self._when_built = SingleObserver()
        self._when_closed = SingleObserver()

    # XXX backwards-compat for old .is_built for now
    @property
    def is_built(self):
        return self.when_built()

    def when_built(self):
        """
        Returns a Deferred that is callback()'d (with this Circuit
        instance) when this circuit hits BUILT.

        If it's already BUILT when this is called, you get an
        already-successful Deferred; otherwise, the state must change
        to BUILT.

        If the circuit will never hit BUILT (e.g. it is abandoned by
        Tor before it gets to BUILT) you will receive an errback
        """
        # XXX note to self: we never do an errback; fix this behavior
        if self.state == 'BUILT':
            return defer.succeed(self)
        return self._when_built.when_fired()

    def when_closed(self):
        """
        Returns a Deferred that callback()'s (with this Circuit instance)
        when this circuit hits CLOSED or FAILED.
        """
        if self.state in ['CLOSED', 'FAILED']:
            return defer.succeed(self)
        return self._when_closed.when_fired()

    def web_agent(self, reactor, socks_endpoint, pool=None):
        """
        :param socks_endpoint: create one with
            :meth:`txtorcon.TorConfig.create_socks_endpoint`. Can be a
            Deferred.

        :param pool: passed on to the Agent (as ``pool=``)
        """
        # local import because there isn't Agent stuff on some
        # platforms we support, so this will only error if you try
        # this on the wrong platform (pypy [??] and old-twisted)
        from txtorcon import web
        return web.tor_agent(
            reactor,
            socks_endpoint,
            circuit=self,
            pool=pool,
        )

    # XXX should make this API match above web_agent (i.e. pass a
    # socks_endpoint) or change the above...
    def stream_via(self, reactor, host, port,
                   socks_endpoint,
                   use_tls=False):
        """
        This returns an `IStreamClientEndpoint`_ that will connect to
        the given ``host``, ``port`` via Tor -- and via this
        parciular circuit.

        We match the streams up using their source-ports, so even if
        there are many streams in-flight to the same destination they
        will align correctly. For example, to cause a stream to go to
        ``torproject.org:443`` via a particular circuit::

            @inlineCallbacks
            def main(reactor):
                circ = yield torstate.build_circuit()  # lets Tor decide the path
                yield circ.when_built()
                tor_ep = circ.stream_via(reactor, 'torproject.org', 443)
                # 'factory' is for your protocol
                proto = yield tor_ep.connect(factory)

        Note that if you're doing client-side Web requests, you
        probably want to use `treq
        <http://treq.readthedocs.org/en/latest/>`_ or ``Agent``
        directly so call :meth:`txtorcon.Circuit.web_agent` instead.

        :param socks_endpoint: should be a Deferred firing a valid
            IStreamClientEndpoint pointing at a Tor SOCKS port (or an
            IStreamClientEndpoint already).

        .. _istreamclientendpoint: https://twistedmatrix.com/documents/current/api/twisted.internet.interfaces.IStreamClientEndpoint.html
        """
        from .endpoints import TorClientEndpoint
        ep = TorClientEndpoint(
            host, port, socks_endpoint,
            tls=use_tls,
            reactor=reactor,
        )
        return TorCircuitEndpoint(reactor, self._torstate, self, ep)

    @property
    def time_created(self):
        if self._time_created is not None:
            return self._time_created
        if 'TIME_CREATED' in self.flags:
            # strip off milliseconds
            t = self.flags['TIME_CREATED'].split('.')[0]
            tstruct = time.strptime(t, TIME_FORMAT)
            self._time_created = datetime(*tstruct[:7])
        return self._time_created

    def listen(self, listener):
        if listener not in self.listeners:
            self.listeners.append(listener)

    def unlisten(self, listener):
        self.listeners.remove(listener)

    def close(self, **kw):
        """
        This asks Tor to close the underlying circuit object. See
        :meth:`txtorcon.torstate.TorState.close_circuit`
        for details.

        You may pass keyword arguments to take care of any Flags Tor
        accepts for the CLOSECIRCUIT command. Currently, this is only
        "IfUnused". So for example: circ.close(IfUnused=True)

        :return: Deferred which callbacks with this Circuit instance
            ONLY after Tor has confirmed it is gone (not simply that the
            CLOSECIRCUIT command has been queued). This could be a while
            if you included IfUnused.
        """

        # we're already closed; nothing to do
        if self.state == 'CLOSED':
            return defer.succeed(None)

        # someone already called close() but we're not closed yet
        if self._closing_deferred:
            d = defer.Deferred()

            def closed(arg):
                d.callback(arg)
                return arg
            self._closing_deferred.addBoth(closed)
            return d

        # actually-close the circuit
        self._closing_deferred = defer.Deferred()

        def close_command_is_queued(*args):
            return self._closing_deferred
        d = self._torstate.close_circuit(self.id, **kw)
        d.addCallback(close_command_is_queued)
        return d

    def age(self, now=None):
        """
        Returns an integer which is the difference in seconds from
        'now' to when this circuit was created.

        Returns None if there is no created-time.
        """
        if not self.time_created:
            return None
        if now is None:
            now = datetime.utcnow()
        return (now - self.time_created).seconds

    def _create_flags(self, kw):
        """
        this clones the kw dict, adding a lower-case version of every
        key (duplicated in stream.py; put in util?)
        """

        flags = {}
        for k in kw.keys():
            flags[k] = kw[k]
            flags[k.lower()] = kw[k]
        return flags

    def update(self, args):
        # print "Circuit.update:",args
        if self.id is None:
            self.id = int(args[0])
            for x in self.listeners:
                x.circuit_new(self)

        else:
            if int(args[0]) != self.id:
                raise RuntimeError("Update for wrong circuit.")
        self.state = args[1]

        kw = find_keywords(args)
        self.flags = kw
        if 'PURPOSE' in kw:
            self.purpose = kw['PURPOSE']
        if 'BUILD_FLAGS' in kw:
            self.build_flags = kw['BUILD_FLAGS'].split(',')

        if self.state == 'LAUNCHED':
            self.path = []
            for x in self.listeners:
                x.circuit_launched(self)
        else:
            if self.state != 'FAILED' and self.state != 'CLOSED':
                if len(args) > 2:
                    self.update_path(args[2].split(','))

        if self.state == 'BUILT':
            for x in self.listeners:
                x.circuit_built(self)
            self._when_built.fire(self)

        elif self.state == 'CLOSED':
            if len(self.streams) > 0:
                # it seems this can/does happen if a remote router
                # crashes or otherwise shuts down a circuit with
                # streams on it still .. also if e.g. you "carml circ
                # --delete " the circuit while the stream is
                # in-progress...can we do better than logging?
                # *should* we do anything else (the stream should get
                # closed still by Tor).
                log.msg(
                    "Circuit is {} but still has {} streams".format(
                        self.state, len(self.streams)
                    )
                )
            flags = self._create_flags(kw)
            self.maybe_call_closing_deferred()
            for x in self.listeners:
                x.circuit_closed(self, **flags)

        elif self.state == 'FAILED':
            if len(self.streams) > 0:
                log.err(RuntimeError("Circuit is %s but still has %d streams" %
                                     (self.state, len(self.streams))))
            flags = self._create_flags(kw)
            self.maybe_call_closing_deferred()
            for x in self.listeners:
                x.circuit_failed(self, **flags)

    def maybe_call_closing_deferred(self):
        """
        Used internally to callback on the _closing_deferred if it
        exists.
        """

        if self._closing_deferred:
            self._closing_deferred.callback(self)
            self._closing_deferred = None
        self._when_closed.fire(self)

    def update_path(self, path):
        """
        There are EXTENDED messages which don't include any routers at
        all, and any of the EXTENDED messages may have some arbitrary
        flags in them. So far, they're all upper-case and none start
        with $ luckily. The routers in the path should all be
        LongName-style router names (this depends on them starting
        with $).

        For further complication, it's possible to extend a circuit to
        a router which isn't in the consensus. nickm via #tor thought
        this might happen in the case of hidden services choosing a
        rendevouz point not in the current consensus.
        """

        oldpath_len = len(self.path)
        self.path = []
        for p in path:
            if p[0] != '$':
                break

            # this will create a Router if we give it a router
            # LongName that doesn't yet exist
            router = self.router_container.router_from_id(p)

            self.path.append(router)
            # if the path grew, notify listeners
            if len(self.path) > oldpath_len:
                for x in self.listeners:
                    x.circuit_extend(self, router)
                oldpath_len = len(self.path)

    def __str__(self):
        path = ' '.join([x.ip for x in self.path])
        return "<Circuit %d %s [%s] for %s>" % (self.id, self.state, path,
                                                self.purpose) def build_tor_connection(connection, build_state=True, wait_for_proto=True,
                         password_function=lambda: None):
    """
    This is used to build a valid TorState (which has .protocol for
    the TorControlProtocol). For example::

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

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

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

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

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

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

    if IStreamClientEndpoint.providedBy(connection):
        endpoint = connection

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

    d = endpoint.connect(
        TorProtocolFactory(
            password_function=password_function
        )
    )
    if build_state:
        d.addCallback(build_state
                      if isinstance(build_state, Callable)
                      else _build_state)
    elif wait_for_proto:
        d.addCallback(wait_for_proto
                      if isinstance(wait_for_proto, Callable)
                      else _wait_for_proto)
    return d def build_local_tor_connection(reactor, host='127.0.0.1', port=9051,
                               socket='/var/run/tor/control', *args, **kwargs):
    """
    This builds a connection to a local Tor, either via 127.0.0.1:9051
    or /var/run/tor/control (by default; the latter is tried
    first). See also :meth:`build_tor_connection
    <txtorcon.torstate.build_tor_connection>` for other key-word
    arguments that are accepted here also.

    **Note**: new code should use :meth:`txtorcon.connect` instead.

    :param host:
        An IP address to find Tor at. Corresponds to the
        ControlListenAddress torrc option.

    :param port:
        The port to use with the address when trying to contact
        Tor. This corresponds to the ControlPort option in torrc
        (default is 9051).
    """

    try:
        return build_tor_connection((reactor, socket), *args, **kwargs)
    except Exception:
        return build_tor_connection((reactor, host, port), *args, **kwargs) def build_timeout_circuit(tor_state, reactor, path, timeout, using_guards=False):
    """
    Build a new circuit within a timeout.

    CircuitBuildTimedOutError will be raised unless we receive a
    circuit build result (success or failure) within the `timeout`
    duration.

    :returns: a Deferred which fires when the circuit build succeeds (or
        fails to build).
    """
    timed_circuit = []
    d = tor_state.build_circuit(routers=path, using_guards=using_guards)

    def get_circuit(c):
        timed_circuit.append(c)
        return c

    def trap_cancel(f):
        f.trap(defer.CancelledError)
        if timed_circuit:
            d2 = timed_circuit[0].close()
        else:
            d2 = defer.succeed(None)
        d2.addCallback(lambda _: Failure(CircuitBuildTimedOutError("circuit build timed out")))
        return d2

    d.addCallback(get_circuit)
    d.addCallback(lambda circ: circ.when_built())
    d.addErrback(trap_cancel)

    reactor.callLater(timeout, d.cancel)
    return d class CircuitBuildTimedOutError(Exception):
    """
    This exception is thrown when using `timed_circuit_build`
    and the circuit build times-out.
    """ class IStreamAttacher(Interface):
    """
    Used by :class:`txtorcon.TorState` to map streams to circuits (see
    :meth:`txtorcon.TorState.set_attacher`).

    Each time a new :class:`txtorcon.Stream` is created, this
    interface will be queried to find out which
    :class:`txtorcon.Circuit` it should be attached to.

    Only advanced use-cases should need to use this directly; for most
    users, using the :func:`txtorcon.Circuit.stream_via` interface
    should be preferred.
    """

    def attach_stream_failure(stream, fail):
        """
        :param stream:
            The stream we were trying to attach.

        :param fail:
            A Failure instance.

        A failure has occurred while trying to attach the stream.
        """

    def attach_stream(stream, circuits):
        """
        :param stream:
            The stream to attach, which will be in NEW or NEWRESOLVE
            state.

        :param circuits:
            all currently available :class:`txtorcon.Circuit` objects
            in the :class:`txtorcon.TorState` in a dict indexed by id.
            Note they are *not* limited to BUILT circuits.

        You should return a :class:`txtorcon.Circuit` instance which
        should be at state BUILT in the currently running Tor. You may
        also return a Deferred which will callback with the desired
        circuit. In this case, you will probably need to be aware that
        the callback from :meth:`txtorcon.TorState.build_circuit` does
        not wait for the circuit to be in BUILT state.

        Alternatively, you may return None in which case the Tor
        controller will be told to choose a circuit itself.

        Note that Tor will refuse to attach to any circuit not in
        BUILT state; see ATTACHSTREAM in control-spec.txt

        Note also that although you get a request to attach a stream
        that ends in .onion Tor doesn't currently let you specify how
        to attach .onion addresses and will always give a 551 error.
        """ class ICircuitListener(Interface):
    """
    An interface to listen for updates to Circuits.
    """

    def circuit_new(circuit):
        """A new circuit has been created.  You'll always get one of
        these for every Circuit even if it doesn't go through the "launched"
        state."""

    def circuit_launched(circuit):
        "A new circuit has been started."

    def circuit_extend(circuit, router):
        "A circuit has been extended to include a new router hop."

    def circuit_built(circuit):
        """
        A circuit has been extended to all hops (usually 3 for user
        circuits).
        """

    def circuit_closed(circuit, **kw):
        """
        A circuit has been closed cleanly (won't be in controller's list
        any more).

        :param kw:
            A dict of additional args. REASON is alsways included, and
            often REMOTE_REASON also. See the control-spec
            documentation.  As of this writing, REASON is one of the
            following strings: MISC, RESOLVEFAILED, CONNECTREFUSED,
            EXITPOLICY, DESTROY, DONE, TIMEOUT, NOROUTE, HIBERNATING,
            INTERNAL,RESOURCELIMIT, CONNRESET, TORPROTOCOL,
            NOTDIRECTORY, END, PRIVATE_ADDR. However, don't depend on
            that: it could be anything.

            To facilitate declaring args you want in the method
            (e.g. ``circuit_failed(self, circuit, reason=None,
            remote_reason=None, **kw)``) lower-case versions of all the
            keys are also provided (pointing to the same -- usually
            UPPERCASE -- strings as the upper-case keys).

        """

    def circuit_failed(circuit, **kw):
        """
        A circuit has been closed because something went wrong.

        The circuit won't be in the TorState's list anymore.

        :param kw:
            A dict of additional args. REASON is alsways included, and
            often REMOTE_REASON also. See the control-spec
            documentation.  As of this writing, REASON is one of the
            following strings: MISC, RESOLVEFAILED, CONNECTREFUSED,
            EXITPOLICY, DESTROY, DONE, TIMEOUT, NOROUTE, HIBERNATING,
            INTERNAL,RESOURCELIMIT, CONNRESET, TORPROTOCOL,
            NOTDIRECTORY, END, PRIVATE_ADDR. However, don't depend on
            that: it could be anything.

            To facilitate declaring args you want in the method
            (e.g. ``circuit_failed(self, circuit, reason=None,
            remote_reason=None, **kw)``) lower-case versions of all the
            keys are also provided (pointing to the same -- usually
            UPPERCASE -- strings as the upper-case keys).
        """ class IStreamListener(Interface):
    """
    Notifications about changes to a :class:`txtorcon.Stream`.

    If you wish for your listener to be added to *all* new streams,
    see :meth:`txtorcon.TorState.add_stream_listener`.
    """

    def stream_new(stream):
        "a new stream has been created"

    def stream_succeeded(stream):
        "stream has succeeded"

    def stream_attach(stream, circuit):
        "the stream has been attached to a circuit"

    def stream_detach(stream, **kw):
        """
        the stream has been detached from its circuit

        :param kw:
            provides any flags for this event, which will include at
            least REASON (but may include anything). See control-spec.
        """

    def stream_closed(stream, **kw):
        """
        stream has been closed (won't be in controller's list anymore).

        :param kw:
            provides any flags for this event, which will include at
            least REASON (but may include anything). See control-spec.
        """

    def stream_failed(stream, **kw):
        """
        stream failed for some reason (won't be in controller's list anymore).

        :param kw:
            a dict of all the flags for the stream failure; see
            control-spec but these will include REASON and sometimes
            REMOTE_REASON (if the remote Tor closed the
            connection). Both an all-uppercase and all-lowercase
            version of each keyword is supplied (by the library; Tor
            provides all-uppercase only). Others may include
            BUILD_FLAGS, PURPOSE, HS_STATE, REND_QUERY, TIME_CREATED
            (or anything else).
        """ class StreamListenerMixin(object):
    """
    Implements all of :class:`txtorcon.IStreamListener` with no-op
    methods. You may subclass from this if you don't care about most
    of the notifications.
    """

    def stream_new(self, stream):
        pass

    def stream_succeeded(self, stream):
        pass

    def stream_attach(self, stream, circuit):
        pass

    def stream_detach(self, stream, **kw):
        pass

    def stream_closed(self, stream, **kw):
        pass

    def stream_failed(self, stream, **kw):
        pass class CircuitListenerMixin(object):
    """
    Implements all of ICircuitListener with no-op methods. Subclass
    from this if you don't care about most of the notifications.
    """

    def circuit_new(self, circuit):
        pass

    def circuit_launched(self, circuit):
        pass

    def circuit_extend(self, circuit, router):
        pass

    def circuit_built(self, circuit):
        pass

    def circuit_closed(self, circuit, **kw):
        pass

    def circuit_failed(self, circuit, **kw):
        pass @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) def _extract_reason(kw):
    """
    Internal helper. Extracts a reason (possibly both reasons!) from
    the kwargs for a circuit failed or closed event.
    """
    try:
        # we "often" have a REASON
        reason = kw['REASON']
        try:
            # ...and sometimes even have a REMOTE_REASON
            reason = '{}, {}'.format(reason, kw['REMOTE_REASON'])
        except KeyError:
            pass  # should still be the 'REASON' error if we had it
    except KeyError:
        reason = "unknown"
    return reason
from zope.interface import implementer, directlyProvides
from zope.interface.verify import verifyClass
from twisted.trial import unittest
from twisted.test import proto_helpers
from twisted.python.failure import Failure
from twisted.internet import task, defer
from twisted.internet.interfaces import IStreamClientEndpoint, IReactorCore
from ipaddress import IPv4Address
from mock import patch, Mock
from txtorcon import TorControlProtocol
from txtorcon import TorProtocolError
from txtorcon import TorState
from txtorcon import Stream
from txtorcon import Circuit
from txtorcon import build_tor_connection
from txtorcon import build_local_tor_connection
from txtorcon import build_timeout_circuit
from txtorcon import CircuitBuildTimedOutError
from txtorcon.interface import IStreamAttacher
from txtorcon.interface import ICircuitListener
from txtorcon.interface import IStreamListener
from txtorcon.interface import StreamListenerMixin
from txtorcon.interface import CircuitListenerMixin
from txtorcon.circuit import _get_circuit_attacher
from txtorcon.circuit import _extract_reason
    from .py3_torstate import TorStatePy3Tests  # noqa
        from .test_torconfig import FakeControlProtocol
import tempfile
        import ipaddress
    def checker(self, state, stream, arg=None):
        if self.expected[0][0] != state:
            raise RuntimeError(
                'Expected event "%s" not "%s".' % (self.expected[0][0], state)
            )
        for (k, v) in self.expected[0][1].items():
            if k == 'arg':
                if v != arg:
                    raise RuntimeError(
                        'Expected argument to have value "%s", not "%s"' %
                        (arg, v)
                    )
            elif getattr(stream, k) != v:
                raise RuntimeError(
                    'Expected attribute "%s" to have value "%s", not "%s"' %
                    (k, v, getattr(stream, k))
                )
        self.expected = self.expected[1:]

    def stream_new(self, stream):
        self.checker('new', stream)

    def stream_succeeded(self, stream):
        self.checker('succeeded', stream)

    def stream_attach(self, stream, circuit):
        self.checker('attach', stream, circuit)

    def stream_closed(self, stream):
        self.checker('closed', stream)

    def stream_failed(self, stream, reason, remote_reason):
        self.checker('failed', stream, reason)


@implementer(IReactorCore)
class FakeReactor:

    def __init__(self, test):
        self.test = test

    def addSystemEventTrigger(self, *args):
        self.test.assertEqual(args[0], 'before')
        self.test.assertEqual(args[1], 'shutdown')
        self.test.assertEqual(args[2], self.test.state.undo_attacher)
        return 1

    def removeSystemEventTrigger(self, id):
        self.test.assertEqual(id, 1)

    def connectTCP(self, *args, **kw):
        """for testing build_tor_connection"""
        raise RuntimeError('connectTCP: ' + str(args))

    def connectUNIX(self, *args, **kw):
        """for testing build_tor_connection"""
        raise RuntimeError('connectUNIX: ' + str(args))


class FakeCircuit(Circuit):

    def __init__(self, id=-999):
        self.streams = []
        self.id = id
        self.state = 'BOGUS'


@implementer(IStreamClientEndpoint)
class FakeEndpoint:

    def get_info_raw(self, keys):
        ans = '\r\n'.join(map(lambda k: '%s=' % k, keys.split()))
        return defer.succeed(ans)

    def get_info_incremental(self, key, linecb):
        linecb('%s=' % key)
        return defer.succeed('')

    def connect(self, protocol_factory):
        self.proto = TorControlProtocol()
        self.proto.transport = proto_helpers.StringTransport()
        self.proto.get_info_raw = self.get_info_raw
        self.proto.get_info_incremental = self.get_info_incremental
        self.proto._set_valid_events(
            'GUARD STREAM CIRC NS NEWCONSENSUS ORCONN NEWDESC ADDRMAP STATUS_GENERAL'
        )

        return defer.succeed(self.proto)


@implementer(IStreamClientEndpoint)
class FakeEndpointAnswers:

    def __init__(self, answers):
        self.answers = answers
        # since we use pop() we need these to be "backwards"
        self.answers.reverse()

    def add_event_listener(self, *args):
        return defer.succeed(None)

    def get_info_raw(self, keys):
        ans = ''
        for k in keys.split():
            if len(self.answers) == 0:
                raise TorProtocolError(551, "ran out of answers")
            ans += '%s=%s\r\n' % (k, self.answers.pop())
        return ans[:-2]                 # don't want trailing \r\n

    def get_info_incremental(self, key, linecb):
        data = self.answers.pop().split('\n')
        if len(data) == 1:
            linecb('{}={}'.format(key, data[0]))
        else:
            linecb('{}='.format(key))
            for line in data:
                linecb(line)
        return defer.succeed('')

    def connect(self, protocol_factory):

        self.proto = TorControlProtocol()