Examples

The examples are grouped by functionality and serve as mini-HOWTOs – if you have a use-case that is missing, it may be useful to add an example, so please file a bug.

All files are in the examples/ sub-directory and are ready to run, usually with defaults designed to work with Tor Browser Bundle (localhost:9151).

The examples use default_control_port() to determine how to connect which you can override with an environment variable: TX_CONTROL_PORT. So e.g. export TX_CONTROL_PORT=9050 to run the examples again a system-wide Tor daemon.

Web: clients

web_client.py

Download the example.

Uses twisted.web.client to download a Web page using a twisted.web.client.Agent, via any circuit Tor chooses.

# this example shows how to use Twisted's web client with Tor via
# txtorcon

from __future__ import print_function

from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import react
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.web.client import readBody

import txtorcon
from txtorcon.util import default_control_port


@react
@inlineCallbacks
def main(reactor):
    # use port 9051 for system tor instances, or:
    # ep = UNIXClientEndpoint(reactor, '/var/run/tor/control')
    # ep = UNIXClientEndpoint(reactor, '/var/run/tor/control')
    ep = TCP4ClientEndpoint(reactor, '127.0.0.1', default_control_port())
    tor = yield txtorcon.connect(reactor, ep)
    print("Connected to {tor} via localhost:{port}".format(
        tor=tor,
        port=default_control_port(),
    ))

    # create a web.Agent that will talk via Tor. If the socks port
    # given isn't yet configured, this will do so. It may also be
    # None, which means "the first configured SOCKSPort"
    # agent = tor.web_agent(u'9999')
    agent = tor.web_agent()
    uri = b'http://surely-this-has-not-been-registered-and-is-invalid.com'
    uri = b'https://www.torproject.org'
    uri = b'http://timaq4ygg2iegci7.onion/'  # txtorcon documentation
    print("Downloading {}".format(uri))
    resp = yield agent.request(b'GET', uri)

    print("Response has {} bytes".format(resp.length))
    body = yield readBody(resp)
    print("received body ({} bytes)".format(len(body)))
    print("{}\n[...]\n{}\n".format(body[:200], body[-200:]))

web_client_treq.py

Download the example.

Uses treq to download a Web page via Tor.

# just copying over most of "carml checkpypi" because it's a good
# example of "I want a stream over *this* circuit".

from __future__ import print_function

from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import react
from twisted.internet.endpoints import TCP4ClientEndpoint

import txtorcon
from txtorcon.util import default_control_port

try:
    import treq
except ImportError:
    print("To use this example, please install 'treq':")
    print("pip install treq")
    raise SystemExit(1)


@react
@inlineCallbacks
def main(reactor):
    ep = TCP4ClientEndpoint(reactor, '127.0.0.1', default_control_port())
    # ep = UNIXClientEndpoint(reactor, '/var/run/tor/control')
    tor = yield txtorcon.connect(reactor, ep)
    print("Connected:", tor)

    resp = yield treq.get(
        'https://www.torproject.org:443',
        agent=tor.web_agent(),
    )

    print("Retrieving {} bytes".format(resp.length))
    data = yield resp.text()
    print("Got {} bytes:\n{}\n[...]{}".format(
        len(data),
        data[:120],
        data[-120:],
    ))

web_client_custom_circuit.py

Download the example.

Builds a custom circuit, and then uses twisted.web.client to download a Web page using the circuit created.

# this example shows how to use specific circuits over Tor (with
# Twisted's web client or with a custom protocol)
#
# NOTE WELL: this functionality is for advanced use-cases and if you
# do anything "special" to select your circuit hops you risk making it
# easy to de-anonymize this (and all other) Tor circuits.

from __future__ import print_function

from twisted.internet.protocol import Protocol, Factory
from twisted.internet.defer import inlineCallbacks, Deferred
from twisted.internet.task import react
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.web.client import readBody

import txtorcon
from txtorcon.util import default_control_port


@react
@inlineCallbacks
def main(reactor):
    # use port 9051 for system tor instances, or:
    # ep = UNIXClientEndpoint(reactor, '/var/run/tor/control')
    ep = TCP4ClientEndpoint(reactor, '127.0.0.1', default_control_port())
    tor = yield txtorcon.connect(reactor, ep)
    print("Connected:", tor)

    config = yield tor.get_config()
    state = yield tor.create_state()
    socks = config.socks_endpoint(reactor)

    # create a custom circuit; in this case we're just letting Tor
    # decide the path but you *can* select a path (again: for advanced
    # use cases that will probably de-anonymize you)
    circ = yield state.build_circuit()
    print("Building a circuit:", circ)

    # at this point, the circuit will be "under way" but may not yet
    # be in BUILT state -- and hence usable. So, we wait. (Just for
    # demo purposes: the underlying connect will wait too)
    yield circ.when_built()
    print("Circuit is ready:", circ)

    if True:
        # create a web.Agent that will use this circuit (or fail)
        agent = circ.web_agent(reactor, socks)

        uri = 'https://www.torproject.org'
        print("Downloading {}".format(uri))
        resp = yield agent.request('GET', uri)

        print("Response has {} bytes".format(resp.length))
        body = yield readBody(resp)
        print("received body ({} bytes)".format(len(body)))
        print("{}\n[...]\n{}\n".format(body[:200], body[-200:]))

    if True:
        # make a plain TCP connection to a thing
        ep = circ.stream_via(reactor, 'torproject.org', 80, config.socks_endpoint(reactor))

        d = Deferred()

        class ToyWebRequestProtocol(Protocol):

            def connectionMade(self):
                print("Connected via {}".format(self.transport.getHost()))
                self.transport.write(
                    'GET http://torproject.org/ HTTP/1.1\r\n'
                    'Host: torproject.org\r\n'
                    '\r\n'
                )

            def dataReceived(self, d):
                print("  received {} bytes".format(len(d)))

            def connectionLost(self, reason):
                print("disconnected: {}".format(reason.value))
                d.callback(None)

        yield ep.connect(Factory.forProtocol(ToyWebRequestProtocol))  # returns "proto"
        yield d
        print("All done, closing the circuit")
        yield circ.close()

Starting Tor

launch_tor.py

Download the example. Launch a new Tor instance. This takes care of setting Tor’s notion ownership so that when the control connection goes away the running Tor exits.

from __future__ import print_function

"""
Launch a private Tor instance.
"""

import sys
import txtorcon
from twisted.web.client import readBody
from twisted.internet.task import react
from twisted.internet.defer import inlineCallbacks


@react
@inlineCallbacks
def main(reactor):
    # note that you can pass a few options as kwargs
    # (e.g. data_directory=, or socks_port= ). For other torrc
    # changes, see below.
    tor = yield txtorcon.launch(
        reactor,
        data_directory="./tordata",
        stdout=sys.stdout,
        socks_port='unix:/tmp/tor2/socks',
    )
    # tor = yield txtorcon.connect(
    #     reactor,
    #     clientFromString(reactor, "unix:/var/run/tor/control"),
    # )
    print("Connected to Tor version '{}'".format(tor.protocol.version))

    config = yield tor.get_config()
    state = yield tor.create_state()
    # or state = yield txtorcon.TorState.from_protocol(tor.protocol)

    print("This Tor has PID {}".format(state.tor_pid))
    print("This Tor has the following {} Circuits:".format(len(state.circuits)))
    for c in state.circuits.values():
        print("  {}".format(c))

    endpoint_d = config.socks_endpoint(reactor, u'unix:/tmp/tor2/socks')
    agent = tor.web_agent(socks_endpoint=endpoint_d)
    uri = b'https://www.torproject.org'
    print("Downloading {}".format(uri))
    resp = yield agent.request(b'GET', uri)
    print("Response has {} bytes".format(resp.length))
    body = yield readBody(resp)
    print("received body ({} bytes)".format(len(body)))
    print("{}\n[...]\n{}\n".format(body[:200], body[-200:]))

    # SOCKSPort is 'really' a list of SOCKS ports in Tor now, so we
    # have to set it to a list ... :/
    print("Changing our config (SOCKSPort=9876)")
    # config.SOCKSPort = ['unix:/tmp/foo/bar']
    config.SOCKSPort = ['9876']
    yield config.save()

    print("Querying to see it changed:")
    socksport = yield tor.protocol.get_conf("SOCKSPort")
    print("SOCKSPort", socksport)

launch_tor_endpoint.py

Download the example. Using the txtorcon.TCP4HiddenServiceEndpoint class to start up a Tor with a hidden service pointed to an IStreamServerEndpoint.

from __future__ import print_function

# Here we set up a Twisted Web server and then launch our own tor with
# a configured hidden service directed at the Web server we set
# up. This uses serverFromString to translate the "onion" endpoint
# descriptor into a TCPHiddenServiceEndpoint object...

from twisted.web import server, resource
from twisted.internet.defer import inlineCallbacks
from twisted.internet.task import react, deferLater
from twisted.internet.endpoints import serverFromString

import txtorcon


class Simple(resource.Resource):
    """
    A really simple Web site.
    """
    isLeaf = True

    def render_GET(self, request):
        return "<html>Hello, world! I'm a hidden service!</html>"


@react
@inlineCallbacks
def main(reactor):
    # several ways to proceed here and what they mean:
    #
    # "onion:80":
    #    launch a new Tor instance, configure a hidden service on some
    #    port and pubish descriptor for port 80
    #
    # "onion:80:controlPort=9051:localPort=8080:socksPort=9089:hiddenServiceDir=/home/human/src/txtorcon/hidserv":
    #    connect to existing Tor via control-port 9051, configure a hidden
    #    service listening locally on 8080, publish a descriptor for port
    #    80 and use an explicit hiddenServiceDir (where "hostname" and
    #    "private_key" files are put by Tor). We set SOCKS port
    #    explicitly, too.
    #
    # "onion:80:localPort=8080:socksPort=9089:hiddenServiceDir=/home/human/src/txtorcon/hidserv":
    #    all the same as above, except we launch a new Tor (because no
    #    "controlPort=9051")

    ep = "onion:80:controlPort=9051:localPort=8080:socksPort=9089:hiddenServiceDir=/home/human/src/txtorcon/hidserv"
    ep = "onion:80:localPort=8080:socksPort=9089:hiddenServiceDir=/home/human/src/txtorcon/hidserv"
    ep = "onion:80"
    hs_endpoint = serverFromString(reactor, ep)

    def progress(percent, tag, message):
        bar = int(percent / 10)
        print("[{}{}] {}".format("#" * bar, "." * (10 - bar), message))
    txtorcon.IProgressProvider(hs_endpoint).add_progress_listener(progress)

    # create our Web server and listen on the endpoint; this does the
    # actual launching of (or connecting to) tor.
    site = server.Site(Simple())
    port = yield hs_endpoint.listen(site)
    # XXX new accessor in newer API
    hs = port.onion_service

    # "port" is an IAddress implementor, in this case TorOnionAddress
    # so you can get most useful information from it -- but you can
    # also access .onion_service (see below)
    print(
        "I have set up a hidden service, advertised at:\n"
        "http://{host}:{port}\n"
        "locally listening on {local_address}\n"
        "Will stop in 60 seconds...".format(
            host=port.getHost().onion_uri,  # or hs.hostname
            port=port.public_port,
            # port.local_address will be a twisted.internet.tcp.Port
            # or a twisted.internet.unix.Port -- both have .getHost()
            local_address=port.local_address.getHost(),
        )
    )

    # if you prefer, hs (port.onion_service) is an instance providing
    # IOnionService (there's no way to do authenticated services via
    # endpoints yet, but if there was then this would implement
    # IOnionClients instead)
    print("private key:\n{}".format(hs.private_key))

    def sleep(s):
        return deferLater(reactor, s, lambda: None)

    yield sleep(50)
    for i in range(10):
        print("Stopping in {}...".format(10 - i))
        yield sleep(1)

Circuits and Streams

disallow_streams_by_port.py

Download the example. An example using IStreamAttacher which is very simple and does just what it sounds like: never attaches Streams exiting to a port in the “disallowed” list (it also explicitly closes them). Note that Tor already has this feature; this is just to illustrate how to use IStreamAttacher and that you may close streams.

XXX keep this one?

from __future__ import print_function
#
# This uses a very simple custom txtorcon.IStreamAttacher to disallow
# certain streams based solely on their port; by default it closes
# all streams on port 80 or 25 without ever attaching them to a
# circuit.
#
# For a more complex IStreamAttacher example, see
# attach_streams_by_country.py
#

from twisted.python import log
from twisted.internet.task import react
from twisted.internet.defer import inlineCallbacks, Deferred
from twisted.internet.endpoints import clientFromString
from zope.interface import implementer

import txtorcon


@implementer(txtorcon.IStreamAttacher)
class PortFilterAttacher:

    def __init__(self, state):
        self.state = state
        self.disallow_ports = [80, 25]
        print(
            "Disallowing all streams to ports: {ports}".format(
                ports=",".join(map(str, self.disallow_ports)),
            )
        )

    def attach_stream(self, stream, circuits):
        """
        IStreamAttacher API
        """

        def stream_closed(x):
            print("Stream closed:", x)

        if stream.target_port in self.disallow_ports:
            print(
                "Disallowing {stream} to port {stream.target_port}".format(
                    stream=stream,
                )
            )
            d = self.state.close_stream(stream)
            d.addCallback(stream_closed)
            d.addErrback(log.err)
            return txtorcon.TorState.DO_NOT_ATTACH

        # Ask Tor to assign stream to a circuit by itself
        return None


@react
@inlineCallbacks
def main(reactor):
    control_ep = clientFromString(reactor, "tcp:localhost:9051")
    tor = yield txtorcon.connect(reactor, control_ep)
    print("Connected to a Tor version={version}".format(
        version=tor.protocol.version,
    ))
    state = yield tor.create_state()
    yield state.set_attacher(PortFilterAttacher(state), reactor)

    print("Existing streams:")
    for s in state.streams.values():
        print("  ", s)
    yield Deferred()

stream_circuit_logger.py

Download the example. For listening to changes in the Circuit and State objects, this example is the easiest to understand as it just prints out (some of) the events that happen. Run this, then visit some Web sites via Tor to see what’s going on.

#!/usr/bin/env python

# This uses an IStreamListener and an ICircuitListener to log all
# built circuits and all streams that succeed.

from __future__ import print_function

import sys
from twisted.python import log
from twisted.internet.task import react
from twisted.internet.defer import inlineCallbacks, Deferred
import txtorcon


def log_circuit(circuit):
    path = '->'.join(map(lambda x: str(x.location.countrycode), circuit.path))
    log.msg('Circuit %d (%s) is %s for purpose "%s"' %
            (circuit.id, path, circuit.state, circuit.purpose))


def log_stream(stream):
    circ = ''
    if stream.circuit:
        path = '->'.join(map(lambda x: str(x.location.countrycode), stream.circuit.path))
        circ = ' via circuit %d (%s)' % (stream.circuit.id, path)
    proc = txtorcon.util.process_from_address(
        stream.source_addr,
        stream.source_port,
    )
    if proc:
        proc = ' from process "%s"' % (proc, )

    elif stream.source_addr == '(Tor_internal)':
        proc = ' for Tor internal use'

    else:
        proc = ' from remote "%s:%s"' % (str(stream.source_addr),
                                         str(stream.source_port))
    log.msg('Stream %d to %s:%d attached%s%s' %
            (stream.id, stream.target_host, stream.target_port, circ, proc))


class StreamCircuitLogger(txtorcon.StreamListenerMixin,
                          txtorcon.CircuitListenerMixin):

    def stream_attach(self, stream, circuit):
        log_stream(stream)

    def stream_failed(self, stream, reason='', remote_reason='', **kw):
        print('Stream %d failed because "%s"' % (stream.id, remote_reason))

    def circuit_built(self, circuit):
        log_circuit(circuit)

    def circuit_failed(self, circuit, **kw):
        log.msg('Circuit %d failed "%s"' % (circuit.id, kw['REASON']))


@react
@inlineCallbacks
def main(reactor):
    log.startLogging(sys.stdout)

    tor = yield txtorcon.connect(reactor)
    log.msg('Connected to a Tor version %s' % tor.protocol.version)
    state = yield tor.create_state()

    listener = StreamCircuitLogger()
    state.add_circuit_listener(listener)
    state.add_stream_listener(listener)

    tor.protocol.add_event_listener('STATUS_GENERAL', log.msg)
    tor.protocol.add_event_listener('STATUS_SERVER', log.msg)
    tor.protocol.add_event_listener('STATUS_CLIENT', log.msg)

    log.msg('Existing state when we connected:')
    for s in state.streams.values():
        log_stream(s)

    log.msg('Existing circuits:')
    for c in state.circuits.values():
        log_circuit(c)
    yield Deferred()

Events

monitor.py

Download the example.

Use a plain txtorcon.TorControlProtocol instance to listen for some simple events – in this case marginally useful, as it listens for logging at level INFO, NOTICE, WARN and ERR.

#!/usr/bin/env python

# Just listens for a few EVENTs from Tor (INFO NOTICE WARN ERR) and
# prints out the contents, so functions like a log monitor.

from __future__ import print_function

from twisted.internet import task, defer
from twisted.internet.endpoints import UNIXClientEndpoint
import txtorcon


@task.react
@defer.inlineCallbacks
def main(reactor):
    ep = UNIXClientEndpoint(reactor, '/var/run/tor/control')
    tor = yield txtorcon.connect(reactor, ep)

    def log(msg):
        print(msg)
    print("Connected to a Tor version", tor.protocol.version)
    for event in ['INFO', 'NOTICE', 'WARN', 'ERR']:
        tor.protocol.add_event_listener(event, log)
    is_current = yield tor.protocol.get_info('status/version/current')
    version = yield tor.protocol.get_info('version')
    print("Version '{}', is_current={}".format(version, is_current['status/version/current']))
    yield defer.Deferred()

Miscellaneous

stem_relay_descriptor.py

Download the example.

Get information about a relay descriptor with the help of Stem’s Relay Descriptor class. We need to specify the nickname or the fingerprint to get back the details.

#!/usr/bin/env python

# This shows how to get the detailed information about a
# relay descriptor and parse it into Stem's RelayDescriptor
# class. More about the class can be read from
#
# https://stem.torproject.org/api/descriptor/server_descriptor.html#stem.descriptor.server_descriptor.RelayDescriptor
#
# We need to pass the nickname or the fingerprint of the onion router
# for which we need the the descriptor information.
#
# Also you need to configure Tor to actually download these
# descriptors -- by default Tor only downloads "microdescriptors"
# (whose information is already available live via txtorcon.Router
# instances). Set "UseMicrodescriptors 0" to download "full" descriptors
from __future__ import print_function

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

try:
    from stem.descriptor.server_descriptor import RelayDescriptor
except ImportError:
    print("You must install 'stem' to use this example:")
    print("  pip install stem")
    raise SystemExit(1)


@react
@inlineCallbacks
def main(reactor):
    tor = yield txtorcon.connect(reactor)

    or_nickname = "moria1"
    print("Trying to get decriptor information about '{}'".format(or_nickname))
    # If the fingerprint is used in place of nickname then, desc/id/<OR identity>
    # should be used.
    try:
        descriptor_info = yield tor.protocol.get_info('desc/name/' + or_nickname)
    except txtorcon.TorProtocolError:
        print("No information found. Enable descriptor downloading by setting:")
        print("  UseMicrodescritors 0")
        print("In your torrc")
        raise SystemExit(1)

    descriptor_info = descriptor_info.values()[0]
    relay_info = RelayDescriptor(descriptor_info)
    print("The relay's fingerprint is: {}".format(relay_info.fingerprint))
    print("Time in UTC when the descriptor was made: {}".format(relay_info.published))

txtorcon.tac

Download the example

Create your own twisted Service for deploying using twistd.

import functools
from os.path import dirname
import sys
from tempfile import mkdtemp

import txtorcon

from twisted.application import service, internet
from twisted.internet import reactor
from twisted.internet.endpoints import TCP4ClientEndpoint
from twisted.python import log
from twisted.web import static, server
from zope.interface import implements


class TorService(service.Service):
    implements(service.IService)
    directory = dirname(__file__)
    port = 8080

    def __init__(self):
        self.torfactory = txtorcon.TorProtocolFactory()
        self.connection = TCP4ClientEndpoint(reactor, 'localhost', 9052)
        self.resource = server.Site(static.File(self.directory))

    def startService(self):
        service.Service.startService(self)

        reactor.listenTCP(self.port, self.resource)
        self._bootstrap().addCallback(self._complete)

    def _bootstrap(self):
        self.config = txtorcon.TorConfig()
        self.config.HiddenServices = [
            txtorcon.HiddenService(self.config, mkdtemp(),
                                   ['%d 127.0.0.1:%d' % (80, self.port)])
        ]
        self.config.save()
        return txtorcon.launch_tor(self.config, reactor,
                                   progress_updates=self._updates,
                                   tor_binary='tor')

    def _updates(self, prog, tag, summary):
        log.msg('%d%%: %s' % (prog, summary))

    def _complete(self, proto):
        log.msg(self.config.HiddenServices[0].hostname)

application = service.Application("Txtorcon Application")
torservice = TorService()
torservice.setServiceParent(application)