class TorProtocolFactory(object):
    """
    Builds TorControlProtocol objects. Implements IProtocolFactory for
    Twisted interaction.

    If your running Tor doesn't support COOKIE authentication, then
    you should supply a password callback.
    """

    def __init__(self, password_function=lambda: None):
        """
        Builds protocols to talk to a Tor client on the specified
        address. For example::

            ep = TCP4ClientEndpoint(reactor, "localhost", 9051)
            ep.connect(TorProtocolFactory())
            reactor.run()

        By default, COOKIE authentication is used if
        available.

        :param password_function:
           If supplied, this is a zero-argument method that returns a
           password (or a Deferred). By default, it returns None. This
           is only queried if the Tor we connect to doesn't support
           (or hasn't enabled) COOKIE authentication.
        """
        self.password_function = password_function

    def doStart(self):
        ":api:`twisted.internet.interfaces.IProtocolFactory` API"

    def doStop(self):
        ":api:`twisted.internet.interfaces.IProtocolFactory` API"

    def buildProtocol(self, addr):
        ":api:`twisted.internet.interfaces.IProtocolFactory` API"
        proto = TorControlProtocol(self.password_function)
        proto.factory = self
        return proto 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 _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 class Router(object):
    """
    Represents a Tor Router, including location.

    The controller you pass in is really only used to do get_info
    calls for ip-to-country/IP in case the
    :class:`txtorcon.util.NetLocation` stuff fails to find a country.

    After an .update() call, the id_hex attribute contains a
    hex-encoded long hash (suitable, for example, to use in a
    ``GETINFO ns/id/*`` call).

    After setting the policy property you may call accepts_port() to
    find out if the router will accept a given port. This works with
    the reject or accept based policies.
    """

    def __init__(self, controller):
        self.controller = controller
        self._flags = []
        self.bandwidth = 0
        self.name_is_unique = False
        self.accepted_ports = None
        self.rejected_ports = None
        self.id_hex = None
        self._location = None
        self.from_consensus = False
        self.ip = 'unknown'
        self.ip_v6 = []                 # most routers have no IPv6 addresses

    unique_name = property(lambda x: x.name_is_unique and x.name or x.id_hex)
    "has the hex id if this router's name is not unique, or its name otherwise"

    @property
    def modified(self):
        """
        This is the time of 'the publication time of its most recent
        descriptor' (in UTC).

        See also dir-spec.txt.

        """
        # "... in the form YYYY-MM-DD HH:MM:SS, in UTC"
        if self._modified is None:
            self._modified = datetime.strptime(
                self._modified_unparsed,
                '%Y-%m-%d %H:%M:%S'
            )
        return self._modified

    def update(self, name, idhash, orhash, modified, ip, orport, dirport):
        self.name = name
        self.id_hash = idhash
        self.or_hash = orhash
        # modified is lazy-parsed, approximately doubling router-parsing time
        self._modified_unparsed = modified
        self._modified = None
        self.ip = ip
        self.or_port = orport
        self.dir_port = dirport
        self._location = None

        self.id_hex = hexIdFromHash(self.id_hash)
        # for py3, these should be valid (but *not* py2)
        # assert type(idhash) is not bytes
        # assert type(orhash) is not bytes

    def get_location(self):
        """
        Returns a Deferred that fires with a NetLocation object for this
        router.
        """
        if self._location:
            return succeed(self._location)
        if self.ip != 'unknown':
            self._location = NetLocation(self.ip)
        else:
            self._location = NetLocation(None)
        if not self._location.countrycode and self.ip != 'unknown':
            # see if Tor is magic and knows more...
            d = self.controller.get_info_raw('ip-to-country/' + self.ip)
            d.addCallback(self._set_country)
            d.addCallback(lambda _: self._location)
            return d
        return succeed(self._location)

    @property
    @deprecated(_Version("txtorcon", 18, 0, 0))
    def location(self):
        """
        A NetLocation instance with some GeoIP or pygeoip information
        about location, asn, city (if available).
        """
        if self._location:
            return self._location

        if self.ip != 'unknown':
            self._location = NetLocation(self.ip)
        else:
            self._location = NetLocation(None)
        if not self._location.countrycode and self.ip != 'unknown':
            # see if Tor is magic and knows more...
            d = self.controller.get_info_raw('ip-to-country/' + self.ip)
            d.addCallback(self._set_country)
            # ignore errors (e.g. "GeoIP Information not loaded")
            d.addErrback(lambda _: None)
        return self._location

    @property
    def flags(self):
        """
        A list of all the flags for this Router, each one an
        all-lower-case string.
        """
        return self._flags

    @flags.setter
    def flags(self, flags):
        """
        It might be nice to make flags not a list of strings. This is
        made harder by the control-spec: `...controllers MUST tolerate
        unrecognized flags and lines...`

        There is some current work in Twisted for open-ended constants
        (enums) support however, it seems.
        """
        if isinstance(flags, (six.text_type, bytes)):
            flags = flags.split()
        self._flags = [x.lower() for x in flags]
        self.name_is_unique = 'named' in self._flags

    @property
    def bandwidth(self):
        """The reported bandwidth of this Router."""
        return self._bandwidth

    @bandwidth.setter
    def bandwidth(self, bw):
        self._bandwidth = int(bw)

    @inlineCallbacks
    def get_onionoo_details(self, agent):
        """
        Requests the 'details' document from onionoo.torproject.org via
        the given `twisted.web.iweb.IAgent` -- you can get a suitable
        instance to pass here by calling either :meth:`txtorcon.Tor.web_agent` or
        :meth:`txtorcon.Circuit.web_agent`.
        """

        # clearnet: 'https://onionoo.torproject.org/details?lookup={}'
        uri = 'http://tgel7v4rpcllsrk2.onion/details?lookup={}'.format(self.id_hex[1:]).encode('ascii')

        resp = yield agent.request(b'GET', uri)
        if resp.code != 200:
            raise RuntimeError(
                'Failed to lookup relay details for {}'.format(self.id_hex)
            )
        body = yield readBody(resp)
        data = json.loads(body.decode('ascii'))
        if len(data['relays']) != 1:
            raise RuntimeError(
                'Got multiple relays for {}'.format(self.id_hex)
            )
        relay_data = data['relays'][0]
        if relay_data['fingerprint'].lower() != self.id_hex[1:].lower():
            raise RuntimeError(
                'Expected "{}" but got data for "{}"'.format(self.id_hex, relay_data['fingerprint'])
            )
        returnValue(relay_data)

    # note that exit-policy is no longer included in the
    # microdescriptors by default, so this stuff is mostly here as a
    # historic artifact. If you want to use exit-policy for things
    # your best bet is to tell your tor to download full descriptors
    # (SETCONF UseMicrodescriptors 0) instead.

    @property
    def policy(self):
        """
        Port policies for this Router.
        :return: a string describing the policy
        """
        if self.accepted_ports:
            return 'accept ' + ','.join(map(str, self.accepted_ports))
        elif self.rejected_ports:
            return 'reject ' + ','.join(map(str, self.rejected_ports))
        else:
            return ''

    @policy.setter
    def policy(self, args):
        """
        setter for the policy descriptor
        """

        word = args[0]
        if word == 'reject':
            self.accepted_ports = None
            self.rejected_ports = []
            target = self.rejected_ports

        elif word == 'accept':
            self.accepted_ports = []
            self.rejected_ports = None
            target = self.accepted_ports

        else:
            raise RuntimeError("Don't understand policy word \"%s\"" % word)

        for port in args[1].split(','):
            if '-' in port:
                (a, b) = port.split('-')
                target.append(PortRange(int(a), int(b)))
            else:
                target.append(int(port))

    def accepts_port(self, port):
        """
        Query whether this Router will accept the given port.
        """

        if self.rejected_ports is None and self.accepted_ports is None:
            raise RuntimeError("policy hasn't been set yet")

        if self.rejected_ports:
            for x in self.rejected_ports:
                if port == x:
                    return False
            return True

        for x in self.accepted_ports:
            if port == x:
                return True
        return False

    def _set_country(self, c):
        """
        callback if we used Tor's GETINFO ip-to-country
        """

        self.location.countrycode = c.split()[0].split('=')[1].strip().upper()

    def __repr__(self):
        n = self.id_hex
        if self.name_is_unique:
            n = self.name
        return "<Router %s %s %s>" % (n, self.location.countrycode,
                                      self.policy) def hashFromHexId(hexid):
    """
    From a hex fingerprint id, convert back to base-64 encoded value.
    """
    if hexid[0] == '$':
        hexid = hexid[1:]
    return b64encode(a2b_hex(hexid))[:-1].decode('ascii') def parse_keywords(lines, multiline_values=True, key_hints=None):
    """
    Utility method to parse name=value pairs (GETINFO etc). Takes a
    string with newline-separated lines and expects at most one = sign
    per line. Accumulates multi-line values.

    :param multiline_values:
        The default is True which allows for multi-line values until a
        line with the next = sign on it. So: '''Foo=bar\nBar'''
        produces one key, 'Foo', with value 'bar\nBar' -- set to
        False, there would be two keys: 'Foo' with value 'bar' and
        'Bar' with value DEFAULT_VALUE.
    """

    rtn = {}
    key = None
    value = ''
    # FIXME could use some refactoring to reduce code duplication!
    for line in lines.split('\n'):
        if line.strip() == 'OK':
            continue

        sp = line.split('=', 1)
        found_key = ('=' in line and ' ' not in sp[0])
        if found_key and key_hints and sp[0] not in key_hints:
            found_key = False
        if found_key:
            if key:
                if key in rtn:
                    if isinstance(rtn[key], list):
                        rtn[key].append(unquote(value))
                    else:
                        rtn[key] = [rtn[key], unquote(value)]
                else:
                    rtn[key] = unquote(value)
            (key, value) = line.split('=', 1)

        else:
            if key is None:
                rtn[line.strip()] = DEFAULT_VALUE

            elif multiline_values is False:
                rtn[key] = value
                rtn[line.strip()] = DEFAULT_VALUE
                key = None
                value = ''

            else:
                value = value + '\n' + line
    if key:
        if key in rtn:
            if isinstance(rtn[key], list):
                rtn[key].append(unquote(value))
            else:
                rtn[key] = [rtn[key], unquote(value)]
        else:
            rtn[key] = unquote(value)
    return rtn 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 ITorControlProtocol(Interface):
    """
    This defines the API to the TorController object.

    This is the usual entry-point to this library, and you shouldn't
    need to call methods outside this interface.
    """

    def get_info(info):
        """
        :return: a Deferred which will callback with the info keys you
           asked for. For values ones, see control-spec.
        """

    def get_conf(*args):
        """
        Returns one or many configuration values via Deferred. See
        control-spec for valid keys. The value will be a dictionary.
        """

    def signal(signal_name):
        """
        Issues a signal to Tor. See control-spec or .valid_signals for
        which ones are available and their return values.
        """

    def build_circuit(routers):
        """
        Builds a circuit consisting of exactly the routers specified,
        in order.  This issues a series of EXTENDCIRCUIT calls to Tor;
        the deferred returned from this is for the final
        EXTEND. FIXME: should return the Circuit instance, but
        currently returns final extend message 'EXTEND 1234' for
        example.
        """

    def close_circuit(circuit):
        """
        Asks Tor to close the circuit. Note that the Circuit instance
        is only removed as a result of the next CIRC CLOSED event. The
        Deferred returned from this method callbacks when the
        CLOSECIRCUIT command has successfully executed, not when the
        circuit is actually gone.

        If you wish to know when this circuit is actually gone, add an
        ICircuitListener and wait for circuit_closed()
        """

    def add_event_listener(evt, callback):
        """
        Add a listener to an Event object. This may be called multiple
        times for the same event. Every time the event happens, the
        callback method will be called. The callback has one argument
        (a string, the contents of the event, minus the '650' and the
        name of the event)

        FIXME: should have an interface for the callback.
        """ class IRouterContainer(Interface):

    unique_routers = Attribute("contains a list of all the Router instances")

    def router_from_id(routerid):
        """
        Note that this method MUST always return a Router instance --
        if you ask for a router ID that didn't yet exist, it is
        created (although without IP addresses and such because it
        wasn't in the consensus). You may find out if a Router came
        from the 'GETINFO ns/all' list by checking the from_consensus
        attribute. This is to simplify code like in Circuit.update()
        that needs to handle the case where an EXTENDED circuit event
        is the only time we've seen a Router -- it's possible for Tor
        to do things with routers not in the consensus (like extend
        circuits to them).

        :return: a router by its ID.
        """ 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 ICircuitContainer(Interface):
    """
    An interface that contains a bunch of Circuit objects and can look
    them up by id.
    """

    def find_circuit(circ_id):
        ":return: a circuit for the cird_id, or exception."

    def close_circuit(circuit, **kwargs):
        """
        Close a circuit.
        :return: a Deferred which callbacks when the closing process
        is started (not necessarily finished inside Tor).
        """

    # FIXME do we need an IStreamContainer that Stream instances get?
    # (Currently, they get an ICircuitContainer...)
    def close_stream(stream, **kwargs):
        """
        Close a stream.
        :return: a Deferred which callbacks when the closing process
        is started (not necessarily finished inside Tor).
        """ 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 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 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 class MicrodescriptorParser(object):
    """
    Parsers microdescriptors line by line. New relays are emitted via
    the 'create_relay' callback.
    """

    def __init__(self, create_relay):
        self._create_relay = create_relay
        self._relay_attrs = None

        class die(object):
            __name__ = 'die'  # FIXME? just to ease spagetti.py:82's pain

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

            def __call__(self, *args):
                raise RuntimeError(self.msg % tuple(args))

        waiting_r = State("waiting_r")
        waiting_w = State("waiting_w")
        waiting_p = State("waiting_p")
        waiting_s = State("waiting_s")

        def ignorable_line(x):
            x = x.strip()
            return x in ['.', 'OK', ''] or x.startswith('ns/')

        waiting_r.add_transition(Transition(waiting_r, ignorable_line, None))
        waiting_r.add_transition(Transition(waiting_s, lambda x: x.startswith('r '), self._router_begin))
        # FIXME use better method/func than die!!
        waiting_r.add_transition(Transition(waiting_r, lambda x: not x.startswith('r '), die('Expected "r " while parsing routers not "%s"')))

        waiting_s.add_transition(Transition(waiting_w, lambda x: x.startswith('s '), self._router_flags))
        waiting_s.add_transition(Transition(waiting_s, lambda x: x.startswith('a '), self._router_address))
        waiting_s.add_transition(Transition(waiting_r, ignorable_line, None))
        waiting_s.add_transition(Transition(waiting_r, lambda x: not x.startswith('s ') and not x.startswith('a '), die('Expected "s " while parsing routers not "%s"')))
        waiting_s.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', None))

        waiting_w.add_transition(Transition(waiting_p, lambda x: x.startswith('w '), self._router_bandwidth))
        waiting_w.add_transition(Transition(waiting_r, ignorable_line, None))
        waiting_w.add_transition(Transition(waiting_s, lambda x: x.startswith('r '), self._router_begin))  # "w" lines are optional
        waiting_w.add_transition(Transition(waiting_r, lambda x: not x.startswith('w '), die('Expected "w " while parsing routers not "%s"')))
        waiting_w.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', None))

        waiting_p.add_transition(Transition(waiting_r, lambda x: x.startswith('p '), self._router_policy))
        waiting_p.add_transition(Transition(waiting_r, ignorable_line, None))
        waiting_p.add_transition(Transition(waiting_s, lambda x: x.startswith('r '), self._router_begin))  # "p" lines are optional
        waiting_p.add_transition(Transition(waiting_r, lambda x: x[:2] != 'p ', die('Expected "p " while parsing routers not "%s"')))
        waiting_p.add_transition(Transition(waiting_r, lambda x: x.strip() == '.', None))

        self._machine = FSM([waiting_r, waiting_s, waiting_w, waiting_p])
        self._relay_attrs = None

    def feed_line(self, line):
        """
        A line has been received.
        """
        self._machine.process(line)

    def done(self, *args):
        """
        All lines have been fed.
        """
        self._maybe_callback_router()

    def _maybe_callback_router(self):
        if self._relay_attrs is not None:
            self._create_relay(**self._relay_attrs)
            self._relay_attrs = None

    def _router_begin(self, data):
        self._maybe_callback_router()
        args = data.split()[1:]
        self._relay_attrs = dict(
            nickname=args[0],
            idhash=args[1],
            orhash=args[2],
            modified=args[3] + ' ' + args[4],
            ip=args[5],
            orport=args[6],
            dirport=args[7],
        )

    def _router_flags(self, data):
        args = data.split()[1:]
        self._relay_attrs['flags'] = args

    def _router_address(self, data):
        """only for IPv6 addresses"""
        args = data.split()[1:]
        try:
            self._relay_attrs['ip_v6'].extend(args)
        except KeyError:
            self._relay_attrs['ip_v6'] = list(args)

    def _router_bandwidth(self, data):
        args = data.split()[1:]
        kw = find_keywords(args)
        self._relay_attrs['bandwidth'] = kw['Bandwidth']

    def _router_policy(self, data):
        pass def hexIdFromHash(thehash):
    """
    From the base-64 encoded hashes Tor uses, this produces the longer
    hex-encoded hashes.

    :param thehash: base64-encoded str
    :return: hex-encoded hash
    """
    return '$' + b2a_hex(b64decode(thehash + '=')).decode('ascii').upper() def maybe_coroutine(obj):
    """
    If 'obj' is a coroutine and we're using Python3, wrap it in
    ensureDeferred. Otherwise return the original object.

    (This is to insert in all callback chains from user code, in case
    that user code is Python3 and used 'async def')
    """
    if six.PY3 and asyncio.iscoroutine(obj):
        return defer.ensureDeferred(obj)
    return obj
import os
import stat
import warnings
    from collections.abc import Callable
    from collections import Callable
from twisted.internet import defer
from twisted.python.failure import Failure
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.internet.endpoints import UNIXClientEndpoint
from twisted.internet.interfaces import IReactorCore
from twisted.internet.interfaces import IStreamClientEndpoint
from zope.interface import implementer
from txtorcon.torcontrolprotocol import TorProtocolFactory
from txtorcon.stream import Stream
from txtorcon.circuit import Circuit, _extract_reason
from txtorcon.router import Router, hashFromHexId
from txtorcon.addrmap import AddrMap
from txtorcon.torcontrolprotocol import parse_keywords
from txtorcon.log import txtorlog
from txtorcon.torcontrolprotocol import TorProtocolError
from txtorcon.interface import ITorControlProtocol
from txtorcon.interface import IRouterContainer
from txtorcon.interface import ICircuitListener
from txtorcon.interface import ICircuitContainer
from txtorcon.interface import IStreamListener
from txtorcon.interface import IStreamAttacher
from txtorcon.interface import StreamListenerMixin
from txtorcon.interface import CircuitListenerMixin
from ._microdesc_parser import MicrodescriptorParser
from .router import hexIdFromHash
from .util import maybe_coroutine
        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()