Merge branch 'tor': preliminary Tor support

setup.py

@@ -22,6 +22,7 @@ setup(name="magic-wormhole",
                     ["wormhole = wormhole.scripts.runner:entry"]},
       install_requires=["spake2==0.3", "pynacl", "requests", "argparse",
                         "six"],
+      extras_require={"tor": ["txtorcon", "ipaddr"]},
       # for Twisted support, we want Twisted>=15.5.0. Older Twisteds don't
       # provide sufficient python3 compatibility.
       test_suite="wormhole.test",

src/wormhole/scripts/cli_args.py

@@ -31,6 +31,8 @@ g.add_argument("--twisted", action="store_true",
                help="use Twisted-based implementations, for testing")
 g.add_argument("--no-listen", action="store_true",
                help="(debug) don't open a listening socket for Transit")
+g.add_argument("--tor", action="store_true",
+               help="use Tor when connecting")
 parser.set_defaults(timing=None)
 subparsers = parser.add_subparsers(title="subcommands",
                                    dest="subcommand")

src/wormhole/scripts/cmd_receive_twisted.py

@@ -39,14 +39,27 @@ class TwistedReceiver(BlockingReceiver):
     # TODO: @handle_server_error
     @inlineCallbacks
     def go(self):
-        w = Wormhole(APPID, self.args.relay_url, timing=self.args.timing)
+        tor_manager = None
+        if self.args.tor:
+            _start = self.args.timing.add_event("import TorManager")
+            from ..twisted.tor_manager import TorManager
+            self.args.timing.finish_event(_start)
+            tor_manager = TorManager(reactor, timing=self.args.timing)
+            # For now, block everything until Tor has started. Soon: launch
+            # tor in parallel with everything else, make sure the TorManager
+            # can lazy-provide an endpoint, and overlap the startup process
+            # with the user handing off the wormhole code
+            yield tor_manager.start()
 
-        rc = yield self._go(w)
+        w = Wormhole(APPID, self.args.relay_url, tor_manager,
+                     timing=self.args.timing)
+
+        rc = yield self._go(w, tor_manager)
         yield w.close()
         returnValue(rc)
 
     @inlineCallbacks
-    def _go(self, w):
+    def _go(self, w, tor_manager):
         self.handle_code(w)
         verifier = yield w.get_verifier()
         self.show_verifier(verifier)
@@ -57,13 +70,13 @@ class TwistedReceiver(BlockingReceiver):
                 returnValue(0)
             if "file" in them_d:
                 f = self.handle_file(them_d)
-                rp = yield self.establish_transit(w, them_d)
+                rp = yield self.establish_transit(w, them_d, tor_manager)
                 yield self.transfer_data(rp, f)
                 self.write_file(f)
                 yield self.close_transit(rp)
             elif "directory" in them_d:
                 f = self.handle_directory(them_d)
-                rp = yield self.establish_transit(w, them_d)
+                rp = yield self.establish_transit(w, them_d, tor_manager)
                 yield self.transfer_data(rp, f)
                 self.write_directory(f)
                 yield self.close_transit(rp)
@@ -96,10 +109,11 @@ class TwistedReceiver(BlockingReceiver):
         yield w.send_data(data)
 
     @inlineCallbacks
-    def establish_transit(self, w, them_d):
+    def establish_transit(self, w, them_d, tor_manager):
         transit_key = w.derive_key(APPID+u"/transit-key")
         transit_receiver = TransitReceiver(self.args.transit_helper,
                                            no_listen=self.args.no_listen,
+                                           tor_manager=tor_manager,
                                            timing=self.args.timing)
         transit_receiver.set_transit_key(transit_key)
         direct_hints = yield transit_receiver.get_direct_hints()

src/wormhole/scripts/cmd_send_twisted.py

@@ -42,11 +42,22 @@ def send_twisted(args):
     print(u"On the other computer, please run: %s" % other_cmd,
           file=args.stdout)
 
-    w = Wormhole(APPID, args.relay_url, timing=args.timing)
+    tor_manager = None
+    if args.tor:
+        from ..twisted.tor_manager import TorManager
+        tor_manager = TorManager(reactor, timing=args.timing)
+        # For now, block everything until Tor has started. Soon: launch tor
+        # in parallel with everything else, make sure the TorManager can
+        # lazy-provide an endpoint, and overlap the startup process with the
+        # user handing off the wormhole code
+        yield tor_manager.start()
+
+    w = Wormhole(APPID, args.relay_url, tor_manager, timing=args.timing)
 
     if fd_to_send:
         transit_sender = TransitSender(args.transit_helper,
                                        no_listen=args.no_listen,
+                                       tor_manager=tor_manager,
                                        timing=args.timing)
         phase1["transit"] = transit_data = {}
         transit_data["relay_connection_hints"] = transit_sender.get_relay_hints()

src/wormhole/scripts/runner.py

@@ -21,6 +21,8 @@ def dispatch(args):
         from ..servers import cmd_usage
         return cmd_usage.tail_usage(args)
 
+    if args.tor:
+        args.twisted = True
     if args.func == "send/send":
         if args.twisted:
             from . import cmd_send_twisted
@@ -29,7 +31,9 @@ def dispatch(args):
         return cmd_send_blocking.send_blocking(args)
     if args.func == "receive/receive":
         if args.twisted:
+            _start = args.timing.add_event("import c_r_t")
             from . import cmd_receive_twisted
+            args.timing.finish_event(_start)
             return cmd_receive_twisted.receive_twisted_sync(args)
         from . import cmd_receive_blocking
         return cmd_receive_blocking.receive_blocking(args)

src/wormhole/twisted/tor_manager.py

@@ -0,0 +1,176 @@
+from __future__ import print_function
+import time
+from zope.interface import implementer
+from twisted.web import error as web_error
+from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.python.compat import nativeString
+from twisted.internet.error import ConnectError
+from twisted.web import iweb
+import txtorcon
+import ipaddr
+from ..timing import DebugTiming
+from .transit import allocate_tcp_port
+
+# based on twisted.web.client._StandardEndpointFactory
+@implementer(iweb.IAgentEndpointFactory)
+class TorWebAgentEndpointFactory(object):
+    def __init__(self, reactor, socks_port):
+        self._reactor = reactor
+        self._socks_port = socks_port
+
+    def endpointForURI(self, uri):
+        try:
+            host = nativeString(uri.host)
+        except UnicodeDecodeError:
+            raise ValueError(("The host of the provided URI ({uri.host!r}) "
+                              "contains non-ASCII octets, it should be ASCII "
+                              "decodable.").format(uri=uri))
+
+        if uri.scheme == b'http':
+            print("building URI endpoint with tor for %s" % uri.toBytes())
+            return txtorcon.TorClientEndpoint(#self._reactor,
+                host, uri.port,
+                socks_hostname="127.0.0.1", socks_port=self._socks_port)
+        elif uri.scheme == b'https':
+            raise NotImplementedError
+            # find some twisted thing that wraps a normal
+            # IStreamClientEndpoint in a TLS-ifying layer, and wrap it around
+            # a TorClientEndpoint. Maybe t.i.endpoints.wrapClientTLS
+        else:
+            raise web_error.SchemeNotSupported("Unsupported scheme: %r" % (uri.scheme,))
+
+class TorManager:
+    def __init__(self, reactor, tor_socks_port=None, tor_control_port=9051,
+                 timing=None):
+        """
+        If tor_socks_port= is provided, I will assume that it points to a
+        functioning SOCKS server, and will use it for all outbound
+        connections. I will not attempt to establish a control-port
+        connection, and I will not be able to run a server.
+
+        Otherwise, I will try to connect to an existing Tor process, first on
+        localhost:9051, then /var/run/tor/control. Then I will try to
+        authenticate, by reading a cookie file named by the Tor process. This
+        will succeed if 1: Tor is already running, and 2: the current user
+        can read that file (either they started it, e.g. TorBrowser, or they
+        are in a unix group that's been given access, e.g. debian-tor).
+
+        If tor_control_port= is provided, I will use it instead of 9051.
+        """
+        self._reactor = reactor
+        # note: False is int
+        assert isinstance(tor_socks_port, (int, type(None)))
+        assert isinstance(tor_control_port, int)
+        self._tor_socks_port = tor_socks_port
+        self._tor_control_port = tor_control_port
+        self._timing = timing or DebugTiming()
+
+    @inlineCallbacks
+    def start(self):
+        # Connect to an existing Tor, or create a new one. If we need to
+        # launch an onion service, then we need a working control port (and
+        # authentication cookie). If we're only acting as a client, we don't
+        # need the control port.
+
+        if self._tor_socks_port is not None:
+            self._can_run_service = False
+            returnValue(True)
+
+        _start_find = self._timing.add_event("find tor")
+        # try port 9051, then try /var/run/tor/control . Throws on failure.
+        state = None
+        _start_tcp = self._timing.add_event("tor localhost")
+        try:
+            connection = (self._reactor, "127.0.0.1", self._tor_control_port)
+            state = yield txtorcon.build_tor_connection(connection)
+            self._tor_protocol = state.protocol
+        except ConnectError:
+            print("unable to reach Tor on %d" % self._tor_control_port)
+            pass
+        self._timing.finish_event(_start_tcp)
+
+        if not state:
+            _start_unix = self._timing.add_event("tor unix")
+            try:
+                connection = (self._reactor, "/var/run/tor/control")
+                # add build_state=False to get back a Protocol object instead
+                # of a State object
+                state = yield txtorcon.build_tor_connection(connection)
+                self._tor_protocol = state.protocol
+            except (ValueError, ConnectError):
+                print("unable to reach Tor on /var/run/tor/control")
+                pass
+            self._timing.finish_event(_start_unix)
+
+        if state:
+            print("connected to pre-existing Tor process")
+            print("state:", state)
+        else:
+            print("launching my own Tor process")
+            yield self._create_my_own_tor()
+            # that sets self._tor_socks_port and self._tor_protocol
+
+        self._timing.finish_event(_start_find)
+        self._can_run_service = True
+        returnValue(True)
+
+    @inlineCallbacks
+    def _create_my_own_tor(self):
+        _start_launch = self._timing.add_event("launch tor")
+        start = time.time()
+        config = self.config = txtorcon.TorConfig()
+        if 0:
+            # The default is for launch_tor to create a tempdir itself, and
+            # delete it when done. We only need to set a DataDirectory if we
+            # want it to be persistent.
+            import tempfile
+            datadir = tempfile.mkdtemp()
+            config.DataDirectory = datadir
+
+        #config.ControlPort = allocate_tcp_port() # defaults to 9052
+        #print("setting config.ControlPort to", config.ControlPort)
+        config.SocksPort = allocate_tcp_port()
+        self._tor_socks_port = config.SocksPort
+        print("setting config.SocksPort to", config.SocksPort)
+
+        tpp = yield txtorcon.launch_tor(config, self._reactor,
+                                        #tor_binary=
+                                        )
+        # gives a TorProcessProtocol with .tor_protocol
+        self._tor_protocol = tpp.tor_protocol
+        print("tp:", self._tor_protocol)
+        print("elapsed:", time.time() - start)
+        self._timing.finish_event(_start_launch)
+        returnValue(True)
+
+    def get_web_agent_endpoint_factory(self):
+        return TorWebAgentEndpointFactory(self._reactor, self._tor_socks_port)
+
+    def is_non_public_numeric_address(self, host):
+        # for numeric hostnames, skip RFC1918 addresses, since no Tor exit
+        # node will be able to reach those. Likewise ignore IPv6 addresses.
+        try:
+            a = ipaddr.IPAddress(host)
+        except ValueError:
+            return False # non-numeric, let Tor try it
+        if a.version != 4:
+            return True # IPv6 gets ignored
+        if (a.is_loopback or a.is_multicast or a.is_private or a.is_reserved
+            or a.is_unspecified):
+            return True # too weird, don't connect
+        return False
+
+    def get_endpoint_for(self, host, port):
+        assert isinstance(port, int)
+        if self.is_non_public_numeric_address(host):
+            print("ignoring non-Tor-able %s" % host)
+            return None
+
+        # txsocksx doesn't like unicode: it concatenates some binary protocol
+        # bytes with the hostname when talking to the SOCKS server, so the
+        # py2 automatic unicode promotion blows up
+        host = host.encode("ascii")
+        ep = txtorcon.TorClientEndpoint(host, port,
+                                        socks_hostname="127.0.0.1",
+                                        socks_port=self._tor_socks_port)
+        return ep

src/wormhole/twisted/transcribe.py

@@ -181,15 +181,23 @@ class Channel:
         return d
 
 class ChannelManager:
-    def __init__(self, relay, appid, side, handle_welcome, timing=None):
+    def __init__(self, relay, appid, side, handle_welcome, tor_manager=None,
+                 timing=None):
         assert isinstance(relay, type(u""))
         self._relay = relay
         self._appid = appid
         self._side = side
         self._handle_welcome = handle_welcome
-        self._timing = timing or DebugTiming()
         self._pool = web_client.HTTPConnectionPool(reactor, True) # persistent
-        self._agent = web_client.Agent(reactor, pool=self._pool)
+        if tor_manager:
+            print("ChannelManager using tor")
+            epf = tor_manager.get_web_agent_endpoint_factory()
+            agent = web_client.Agent.usingEndpointFactory(reactor, epf,
+                                                          pool=self._pool)
+        else:
+            agent = web_client.Agent(reactor, pool=self._pool)
+        self._agent = agent
+        self._timing = timing or DebugTiming()
 
     @inlineCallbacks
     def allocate(self):
@@ -250,13 +258,14 @@ class Wormhole:
     version_warning_displayed = False
     _send_confirm = True
 
-    def __init__(self, appid, relay_url, timing=None):
+    def __init__(self, appid, relay_url, tor_manager=None, timing=None):
         if not isinstance(appid, type(u"")): raise TypeError(type(appid))
         if not isinstance(relay_url, type(u"")):
             raise TypeError(type(relay_url))
         if not relay_url.endswith(u"/"): raise UsageError
         self._appid = appid
         self._relay_url = relay_url
+        self._tor_manager = tor_manager
         self._timing = timing or DebugTiming()
         self._set_side(hexlify(os.urandom(5)).decode("ascii"))
         self.code = None
@@ -272,6 +281,7 @@ class Wormhole:
         self._side = side
         self._channel_manager = ChannelManager(self._relay_url, self._appid,
                                                self._side, self.handle_welcome,
+                                               self._tor_manager,
                                                self._timing)
         self._channel = None
 

src/wormhole/twisted/transit.py

@@ -469,7 +469,7 @@ def there_can_be_only_one(contenders):
 class Common:
     RELAY_DELAY = 2.0
 
-    def __init__(self, transit_relay, no_listen=False,
+    def __init__(self, transit_relay, no_listen=False, tor_manager=None,
                  reactor=reactor, timing=None):
         if transit_relay:
             if not isinstance(transit_relay, type(u"")):
@@ -477,6 +477,7 @@ class Common:
             self._transit_relays = [transit_relay]
         else:
             self._transit_relays = []
+        self._tor_manager = tor_manager
         self._transit_key = None
         self._no_listen = no_listen
         self._waiting_for_transit_key = []
@@ -487,7 +488,7 @@ class Common:
         self._timing_started = self._timing.add_event("transit")
 
     def _build_listener(self):
-        if self._no_listen:
+        if self._no_listen or self._tor_manager:
             return ([], None)
         portnum = allocate_tcp_port()
         direct_hints = [u"tcp:%s:%d" % (addr, portnum)
@@ -686,10 +687,17 @@ class Common:
         # TODO: use transit_common.parse_hint_tcp
         if ":" not in hint:
             return None
+        pieces = hint.split(":")
         hint_type = hint.split(":")[0]
+        if hint_type == "tor" and self._tor_manager:
+            return self._tor_manager.get_endpoint_for(pieces[1], int(pieces[2]))
         if hint_type != "tcp":
             return None
         pieces = hint.split(":")
+        if self._tor_manager:
+            # our TorManager will return None for non-public IPv4 addresses
+            # and any IPv6 address
+            return self._tor_manager.get_endpoint_for(pieces[1], int(pieces[2]))
         return endpoints.HostnameEndpoint(self._reactor, pieces[1],
                                           int(pieces[2]))