rewrite Tor support (py2 only)

The new TorManager adds --launch-tor and --tor-control-port= arguments
(requiring the user to explicitly request a new Tor process, if that's what
they want). The default (when --tor is enabled) looks for a control port in
the usual places (/var/run/tor/control, localhost:9051, localhost:9151), then
falls back to hoping there's a SOCKS port in the usual
place (localhost:9050). (closes #64)

The ssh utilities should now accept the same tor arguments as ordinary
send/receive commands. There are now full tests for TorManager, and basic
tests for how send/receive use it. (closes #97)

Note that Tor is only supported on python2.7 for now, since txsocksx (and
therefore txtorcon) doesn't work on py3. You need to do "pip install
magic-wormhole[tor]" to get Tor support, and that will get you an inscrutable
error on py3 (referencing vcversioner, "install_requires must be a string or
list of strings", and "int object not iterable").

To run tests, you must install with the [dev] extra (to get "mock" and other
libraries). Our setup.py only includes "txtorcon" in the [dev] extra when on
py2, not on py3. Unit tests tolerate the lack of txtorcon (they mock out
everything txtorcon would provide), so they should provide the same coverage
on both py2 and py3.

setup.py

@@ -1,4 +1,4 @@
-
+import sys
 from setuptools import setup
 
 import versioneer
@@ -10,6 +10,14 @@ DEV_REQUIREMENTS = [
     "tox",
     "pyflakes",
 ]
+if sys.version_info[0] < 3:
+    # txtorcon is not yet compatible with py3, so we include "txtorcon" in
+    # DEV_REQUIREMENTS under py2 but not under py3. The test suite will skip
+    # the tor tests when txtorcon is not importable. This results in
+    # different wheels when built under py2 vs py3 (with different
+    # extras_require[dev] dependencies), but I think this is ok, since nobody
+    # should be installing with [dev] from a wheel.
+    DEV_REQUIREMENTS.append("txtorcon")
 
 setup(name="magic-wormhole",
       version=versioneer.get_version(),
@@ -45,7 +53,7 @@ setup(name="magic-wormhole",
       extras_require={
           ':sys_platform=="win32"': ["pypiwin32"],
           "tor": ["txtorcon"],
-          "dev": DEV_REQUIREMENTS,
+          "dev": DEV_REQUIREMENTS, # includes txtorcon on py2, but not py3
       },
       test_suite="wormhole.test",
       cmdclass=commands,

src/wormhole/cli/cli.py

@@ -152,6 +152,12 @@ TorArgs = _compose(
     click.option("--tor", is_flag=True, default=False,
                  help="use Tor when connecting",
                  ),
+    click.option("--launch-tor", is_flag=True, default=False,
+                 help="launch Tor, rather than use existing control/socks port",
+                 ),
+    click.option("--tor-control-port", default=None, metavar="ENDPOINT",
+                 help="endpoint descriptor for Tor control port",
+                 ),
 )
 
 # wormhole send (or "wormhole tx")

src/wormhole/cli/cmd_receive.py

@@ -7,7 +7,7 @@ from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.python import log
 from ..wormhole import wormhole
 from ..transit import TransitReceiver
-from ..errors import TransferError, WormholeClosedError
+from ..errors import TransferError, WormholeClosedError, NoTorError
 from ..util import (dict_to_bytes, bytes_to_dict, bytes_to_hexstr,
                     estimate_free_space)
 
@@ -50,7 +50,11 @@ class TwistedReceiver:
             with self.args.timing.add("import", which="tor_manager"):
                 from ..tor_manager import TorManager
             self._tor_manager = TorManager(self._reactor,
+                                           self.args.launch_tor,
+                                           self.args.tor_control_port,
                                            timing=self.args.timing)
+            if not self._tor_manager.tor_available():
+                raise NoTorError()
             # 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

src/wormhole/cli/cmd_send.py

@@ -6,7 +6,7 @@ from twisted.python import log
 from twisted.protocols import basic
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
-from ..errors import TransferError, WormholeClosedError
+from ..errors import TransferError, WormholeClosedError, NoTorError
 from ..wormhole import wormhole
 from ..transit import TransitSender
 from ..util import dict_to_bytes, bytes_to_dict, bytes_to_hexstr
@@ -40,7 +40,12 @@ class Sender:
         if self._args.tor:
             with self._timing.add("import", which="tor_manager"):
                 from ..tor_manager import TorManager
-            self._tor_manager = TorManager(reactor, timing=self._timing)
+            self._tor_manager = TorManager(reactor,
+                                           self._args.launch_tor,
+                                           self._args.tor_control_port,
+                                           timing=self._timing)
+            if not self._tor_manager.tor_available():
+                raise NoTorError()
             # 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

src/wormhole/cli/cmd_ssh.py

@@ -68,6 +68,8 @@ def accept(cfg, reactor=reactor):
         data=cfg.public_key[2],
         code=cfg.code,
         use_tor=cfg.tor,
+        launch_tor=cfg.launch_tor,
+        tor_control_port=cfg.tor_control_port,
     )
     print("Key sent.")
 
@@ -108,6 +110,8 @@ def invite(cfg, reactor=reactor):
         cfg.relay_url,
         None,  # allocate a code for us
         use_tor=cfg.tor,
+        launch_tor=cfg.launch_tor,
+        tor_control_port=cfg.tor_control_port,
         on_code=on_code_created,
     )
 

src/wormhole/test/test_scripts.py

@@ -4,6 +4,7 @@ from humanize import naturalsize
 import mock
 from twisted.trial import unittest
 from twisted.python import procutils, log
+from twisted.internet import defer, endpoints, reactor
 from twisted.internet.utils import getProcessOutputAndValue
 from twisted.internet.defer import gatherResults, inlineCallbacks
 from .. import __version__
@@ -213,6 +214,18 @@ class ScriptVersion(ServerBase, ScriptsBase, unittest.TestCase):
         self.failUnlessEqual(ver.strip(), "magic-wormhole {}".format(__version__))
         self.failUnlessEqual(rc, 0)
 
+class FakeTorManager:
+    # use normal endpoints, but record the fact that we were asked
+    def __init__(self):
+        self.endpoints = []
+    def tor_available(self):
+        return True
+    def start(self):
+        return defer.succeed(None)
+    def get_endpoint_for(self, host, port):
+        self.endpoints.append((host, port))
+        return endpoints.HostnameEndpoint(reactor, host, port)
+
 class PregeneratedCode(ServerBase, ScriptsBase, unittest.TestCase):
     # we need Twisted to run the server, but we run the sender and receiver
     # with deferToThread()
@@ -224,8 +237,11 @@ class PregeneratedCode(ServerBase, ScriptsBase, unittest.TestCase):
 
     @inlineCallbacks
     def _do_test(self, as_subprocess=False,
-                 mode="text", addslash=False, override_filename=False):
+                 mode="text", addslash=False, override_filename=False,
+                 fake_tor=False):
         assert mode in ("text", "file", "directory", "slow-text")
+        if fake_tor:
+            assert not as_subprocess
         send_cfg = config("send")
         recv_cfg = config("receive")
         message = "blah blah blah ponies"
@@ -341,9 +357,26 @@ class PregeneratedCode(ServerBase, ScriptsBase, unittest.TestCase):
                              (send_res, receive_res))
         else:
             send_cfg.cwd = send_dir
+            recv_cfg.cwd = receive_dir
+
+            if fake_tor:
+                send_cfg.tor = True
+                send_cfg.transit_helper = self.transit
+                tx_tm = FakeTorManager()
+                with mock.patch("wormhole.tor_manager.TorManager",
+                                return_value=tx_tm,
+                                ) as mtx_tm:
                     send_d = cmd_send.send(send_cfg)
 
-            recv_cfg.cwd = receive_dir
+                recv_cfg.tor = True
+                recv_cfg.transit_helper = self.transit
+                rx_tm = FakeTorManager()
+                with mock.patch("wormhole.tor_manager.TorManager",
+                                return_value=rx_tm,
+                                ) as mrx_tm:
+                    receive_d = cmd_receive.receive(recv_cfg)
+            else:
+                send_d = cmd_send.send(send_cfg)
                 receive_d = cmd_receive.receive(recv_cfg)
 
             # The sender might fail, leaving the receiver hanging, or vice
@@ -355,6 +388,20 @@ class PregeneratedCode(ServerBase, ScriptsBase, unittest.TestCase):
             else:
                 yield gatherResults([send_d, receive_d], True)
 
+            if fake_tor:
+                expected_endpoints = [("127.0.0.1", self.relayport)]
+                if mode in ("file", "directory"):
+                    expected_endpoints.append(("127.0.0.1", self.transitport))
+                tx_timing = mtx_tm.call_args[1]["timing"]
+                self.assertEqual(tx_tm.endpoints, expected_endpoints)
+                self.assertEqual(mtx_tm.mock_calls,
+                                 [mock.call(reactor, False, None,
+                                            timing=tx_timing)])
+                rx_timing = mrx_tm.call_args[1]["timing"]
+                self.assertEqual(rx_tm.endpoints, expected_endpoints)
+                self.assertEqual(mrx_tm.mock_calls,
+                                 [mock.call(reactor, False, None,
+                                            timing=rx_timing)])
 
             send_stdout = send_cfg.stdout.getvalue()
             send_stderr = send_cfg.stderr.getvalue()
@@ -447,11 +494,15 @@ class PregeneratedCode(ServerBase, ScriptsBase, unittest.TestCase):
         return self._do_test()
     def test_text_subprocess(self):
         return self._do_test(as_subprocess=True)
+    def test_text_tor(self):
+        return self._do_test(fake_tor=True)
 
     def test_file(self):
         return self._do_test(mode="file")
     def test_file_override(self):
         return self._do_test(mode="file", override_filename=True)
+    def test_file_tor(self):
+        return self._do_test(mode="file", fake_tor=True)
 
     def test_directory(self):
         return self._do_test(mode="directory")

src/wormhole/test/test_tor_manager.py

@@ -0,0 +1,337 @@
+from __future__ import print_function, unicode_literals
+import mock, io
+from twisted.trial import unittest
+from twisted.internet import defer
+from twisted.internet.error import ConnectError
+from six import next
+
+from ..tor_manager import TorManager, DEFAULT_VALUE
+
+class Tor(unittest.TestCase):
+    def test_create(self):
+        tm = TorManager(None)
+        del tm
+
+    def test_bad_args(self):
+        e = self.assertRaises(TypeError,
+                              TorManager, None, launch_tor="not boolean")
+        self.assertEqual(str(e), "launch_tor= must be boolean")
+        e = self.assertRaises(TypeError,
+                              TorManager, None, tor_control_port=1234)
+        self.assertEqual(str(e), "tor_control_port= must be str or None")
+        e = self.assertRaises(ValueError,
+                              TorManager, None, launch_tor=True,
+                              tor_control_port="tcp:127.0.0.1:1234")
+        self.assertEqual(str(e),
+                         "cannot combine --launch-tor and --tor-control-port=")
+
+    def test_start_launch_tor(self):
+        reactor = object()
+        stderr = io.StringIO()
+        tm = TorManager(reactor, launch_tor=True, stderr=stderr)
+        dlt_d = defer.Deferred()
+        tm._do_launch_tor = mock.Mock(return_value=dlt_d)
+        tm._try_control_port = mock.Mock()
+        d = tm.start()
+        self.assertNoResult(d)
+        tsep = object()
+        with mock.patch("wormhole.tor_manager.clientFromString",
+                        return_value=tsep) as cfs:
+            dlt_d.callback(("tproto", "tconfig", "socks_desc"))
+            res = self.successResultOf(d)
+            self.assertEqual(res, None)
+            self.assertEqual(tm._tor_protocol, "tproto")
+            self.assertEqual(tm._tor_config, "tconfig")
+            self.assertEqual(tm._tor_socks_endpoint, tsep)
+            self.assertEqual(tm._do_launch_tor.mock_calls, [mock.call()])
+            self.assertEqual(tm._try_control_port.mock_calls, [])
+            self.assertEqual(cfs.mock_calls, [mock.call(reactor, "socks_desc")])
+
+    def test_start_control_port_default_failure(self):
+        reactor = object()
+        stderr = io.StringIO()
+        tm = TorManager(reactor, stderr=stderr)
+        tm._do_launch_tor = mock.Mock()
+        tcp_ds = [defer.Deferred() for i in range(5)]
+        tcp_ds_iter = iter(tcp_ds)
+        attempted_control_ports = []
+        def next_d(control_port):
+            attempted_control_ports.append(control_port)
+            return next(tcp_ds_iter)
+        tm._try_control_port = mock.Mock(side_effect=next_d)
+        d = tm.start()
+        tsep = object()
+        with mock.patch("wormhole.tor_manager.clientFromString",
+                        return_value=tsep) as cfs:
+            self.assertNoResult(d)
+            self.assertEqual(attempted_control_ports,
+                             ["unix:/var/run/tor/control"])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call("unix:/var/run/tor/control")])
+            tcp_ds[0].callback((None, None, None))
+
+            self.assertNoResult(d)
+            self.assertEqual(attempted_control_ports,
+                             ["unix:/var/run/tor/control",
+                              "tcp:127.0.0.1:9051",
+                              ])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call("unix:/var/run/tor/control"),
+                              mock.call("tcp:127.0.0.1:9051"),
+                              ])
+            tcp_ds[1].callback((None, None, None))
+
+            self.assertNoResult(d)
+            self.assertEqual(attempted_control_ports,
+                             ["unix:/var/run/tor/control",
+                              "tcp:127.0.0.1:9051",
+                              "tcp:127.0.0.1:9151",
+                              ])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call("unix:/var/run/tor/control"),
+                              mock.call("tcp:127.0.0.1:9051"),
+                              mock.call("tcp:127.0.0.1:9151"),
+                              ])
+            tcp_ds[2].callback((None, None, None))
+
+            res = self.successResultOf(d)
+            self.assertEqual(res, None)
+            self.assertEqual(tm._tor_protocol, None)
+            self.assertEqual(tm._tor_config, None)
+            self.assertEqual(tm._tor_socks_endpoint, tsep)
+            self.assertEqual(tm._do_launch_tor.mock_calls, [])
+            self.assertEqual(cfs.mock_calls,
+                             [mock.call(reactor, "tcp:127.0.0.1:9050")])
+
+    def test_start_control_port_default(self):
+        reactor = object()
+        stderr = io.StringIO()
+        tm = TorManager(reactor, stderr=stderr)
+        tm._do_launch_tor = mock.Mock()
+        tcp_d = defer.Deferred()
+        # let it succeed on the first try
+        tm._try_control_port = mock.Mock(return_value=tcp_d)
+        d = tm.start()
+        self.assertNoResult(d)
+        tsep = object()
+        with mock.patch("wormhole.tor_manager.clientFromString",
+                        return_value=tsep) as cfs:
+            tcp_d.callback(("tproto", "tconfig", "socks_desc"))
+            res = self.successResultOf(d)
+            self.assertEqual(res, None)
+            self.assertEqual(tm._tor_protocol, "tproto")
+            self.assertEqual(tm._tor_config, "tconfig")
+            self.assertEqual(tm._tor_socks_endpoint, tsep)
+            self.assertEqual(tm._do_launch_tor.mock_calls, [])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call("unix:/var/run/tor/control")])
+            self.assertEqual(cfs.mock_calls, [mock.call(reactor, "socks_desc")])
+
+    def test_start_control_port_non_default_failure(self):
+        reactor = object()
+        my_port = "my_port"
+        stderr = io.StringIO()
+        tm = TorManager(reactor, tor_control_port=my_port, stderr=stderr)
+        tm._do_launch_tor = mock.Mock()
+        tcp_ds = [defer.Deferred() for i in range(5)]
+        tcp_ds_iter = iter(tcp_ds)
+        attempted_control_ports = []
+        def next_d(control_port):
+            attempted_control_ports.append(control_port)
+            return next(tcp_ds_iter)
+        tm._try_control_port = mock.Mock(side_effect=next_d)
+        d = tm.start()
+        tsep = object()
+        with mock.patch("wormhole.tor_manager.clientFromString",
+                        return_value=tsep) as cfs:
+            self.assertNoResult(d)
+            self.assertEqual(attempted_control_ports, [my_port])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call(my_port)])
+            tcp_ds[0].callback((None, None, None))
+
+            res = self.successResultOf(d)
+            self.assertEqual(res, None)
+            self.assertEqual(tm._tor_protocol, None)
+            self.assertEqual(tm._tor_config, None)
+            self.assertEqual(tm._tor_socks_endpoint, tsep)
+            self.assertEqual(tm._do_launch_tor.mock_calls, [])
+            self.assertEqual(cfs.mock_calls,
+                             [mock.call(reactor, "tcp:127.0.0.1:9050")])
+
+    def test_start_control_port_non_default(self):
+        reactor = object()
+        my_port = "my_port"
+        stderr = io.StringIO()
+        tm = TorManager(reactor, tor_control_port=my_port, stderr=stderr)
+        tm._do_launch_tor = mock.Mock()
+        tcp_d = defer.Deferred()
+        tm._try_control_port = mock.Mock(return_value=tcp_d)
+        d = tm.start()
+        self.assertNoResult(d)
+        tsep = object()
+        with mock.patch("wormhole.tor_manager.clientFromString",
+                        return_value=tsep) as cfs:
+            tcp_d.callback(("tproto", "tconfig", "socks_desc"))
+            res = self.successResultOf(d)
+            self.assertEqual(res, None)
+            self.assertEqual(tm._tor_protocol, "tproto")
+            self.assertEqual(tm._tor_config, "tconfig")
+            self.assertEqual(tm._tor_socks_endpoint, tsep)
+            self.assertEqual(tm._do_launch_tor.mock_calls, [])
+            self.assertEqual(tm._try_control_port.mock_calls,
+                             [mock.call(my_port)])
+            self.assertEqual(cfs.mock_calls, [mock.call(reactor, "socks_desc")])
+
+    def test_launch(self):
+        reactor = object()
+        stderr = io.StringIO()
+        tc = mock.Mock()
+        mock_TorConfig = mock.patch("wormhole.tor_manager.TorConfig",
+                                    return_value=tc)
+        lt_d = defer.Deferred()
+        mock_launch_tor = mock.patch("wormhole.tor_manager.launch_tor",
+                                     return_value=lt_d)
+        mock_allocate_tcp_port = mock.patch("wormhole.tor_manager.allocate_tcp_port",
+                                            return_value=12345)
+        mock_clientFromString = mock.patch("wormhole.tor_manager.clientFromString")
+        with mock_TorConfig as mtc:
+            with mock_launch_tor as mlt:
+                with mock_allocate_tcp_port as matp:
+                    with mock_clientFromString as mcfs:
+                        tm = TorManager(reactor, launch_tor=True, stderr=stderr)
+                        d = tm.start()
+                        self.assertNoResult(d)
+                        tp = mock.Mock()
+                        lt_d.callback(tp)
+        res = self.successResultOf(d)
+        self.assertEqual(res, None)
+        self.assertIs(tm._tor_protocol, tp)
+        self.assertIs(tm._tor_config, tc)
+        self.assertEqual(mtc.mock_calls, [mock.call()])
+        self.assertEqual(mlt.mock_calls, [mock.call(tc, reactor)])
+        self.assertEqual(matp.mock_calls, [mock.call()])
+        self.assertEqual(mcfs.mock_calls,
+                         [mock.call(reactor, "tcp:127.0.0.1:12345")])
+
+    def _do_test_try_control_port(self, socks_ports, exp_socks_desc,
+                                  btc_exception=None, tcfp_exception=None):
+        reactor = object()
+        stderr = io.StringIO()
+        ep = object()
+        mock_clientFromString = mock.patch("wormhole.tor_manager.clientFromString",
+                                           return_value=ep)
+        tproto = mock.Mock()
+        btc_d = defer.Deferred()
+        mock_build_tor_connection = mock.patch("wormhole.tor_manager.build_tor_connection", return_value=btc_d)
+        torconfig = mock.Mock()
+        tc = mock.Mock()
+        tc.SocksPort = iter(socks_ports)
+        tc_d = defer.Deferred()
+        torconfig.from_protocol = mock.Mock(return_value=tc_d)
+        mock_torconfig = mock.patch("wormhole.tor_manager.TorConfig", torconfig)
+
+        control_port = object()
+
+        with mock_clientFromString as cfs:
+            with mock_build_tor_connection as btc:
+                with mock_torconfig:
+                    tm = TorManager(reactor, stderr=stderr)
+                    d = tm._try_control_port(control_port)
+                    # waiting in 'tproto = yield build_tor_connection(..)'
+                    self.assertNoResult(d)
+                    self.assertEqual(cfs.mock_calls,
+                                     [mock.call(reactor, control_port)])
+                    self.assertEqual(btc.mock_calls,
+                                     [mock.call(ep, build_state=False)])
+                    self.assertEqual(torconfig.from_protocol.mock_calls, [])
+
+                    btc_d.callback(tproto)
+                    # waiting in 'tconfig = yield TorConfig.from_protocol(..)'
+                    self.assertNoResult(d)
+                    self.assertEqual(torconfig.from_protocol.mock_calls,
+                                     [mock.call(tproto)])
+
+                    tc_d.callback(tc)
+                    res = self.successResultOf(d)
+                    self.assertEqual(res, (tproto, tc, exp_socks_desc))
+
+    def test_try_control_port(self):
+        self._do_test_try_control_port(["1234 ignorestuff",
+                                        "unix:/foo WorldWritable"],
+                                       "tcp:127.0.0.1:1234")
+        self._do_test_try_control_port(["unix:/foo WorldWritable",
+                                        "1234 ignorestuff"],
+                                       "unix:/foo")
+        self._do_test_try_control_port([DEFAULT_VALUE,
+                                        "1234"],
+                                       "tcp:127.0.0.1:9050")
+
+    def _do_test_try_control_port_exception(self, btc_exc=None, tcfp_exc=None):
+        reactor = object()
+        stderr = io.StringIO()
+        ep = object()
+        mock_clientFromString = mock.patch("wormhole.tor_manager.clientFromString",
+                                           return_value=ep)
+        tproto = mock.Mock()
+        btc_d = defer.Deferred()
+        mock_build_tor_connection = mock.patch("wormhole.tor_manager.build_tor_connection", return_value=btc_d)
+        torconfig = mock.Mock()
+        tcfp_d = defer.Deferred()
+        torconfig.from_protocol = mock.Mock(return_value=tcfp_d)
+        mock_torconfig = mock.patch("wormhole.tor_manager.TorConfig", torconfig)
+
+        control_port = object()
+
+        with mock_clientFromString:
+            with mock_build_tor_connection:
+                with mock_torconfig:
+                    tm = TorManager(reactor, stderr=stderr)
+                    d = tm._try_control_port(control_port)
+                    # waiting in 'tproto = yield build_tor_connection(..)'
+                    self.assertNoResult(d)
+
+                    if btc_exc:
+                        btc_d.errback(btc_exc)
+                    else:
+                        btc_d.callback(tproto)
+                        assert tcfp_exc
+                        tcfp_d.errback(tcfp_exc)
+
+                    res = self.successResultOf(d)
+                    self.assertEqual(res, (None, None, None))
+
+    def test_try_control_port_error(self):
+        self._do_test_try_control_port_exception(btc_exc=ValueError())
+        self._do_test_try_control_port_exception(btc_exc=ConnectError())
+        self._do_test_try_control_port_exception(tcfp_exc=ValueError())
+        self._do_test_try_control_port_exception(tcfp_exc=ConnectError())
+
+    def test_badaddr(self):
+        tm = TorManager(None)
+        isnon = tm.is_non_public_numeric_address
+        self.assertTrue(isnon("10.0.0.1"))
+        self.assertTrue(isnon("127.0.0.1"))
+        self.assertTrue(isnon("192.168.78.254"))
+        self.assertTrue(isnon("::1"))
+        self.assertFalse(isnon("8.8.8.8"))
+        self.assertFalse(isnon("example.org"))
+
+    def test_endpoint(self):
+        reactor = object()
+        stderr = io.StringIO()
+        tm = TorManager(reactor, stderr=stderr)
+        tm._tor_socks_endpoint = tse = object()
+        exp_ep = object()
+        with mock.patch("wormhole.tor_manager.TorClientEndpoint",
+                        return_value=exp_ep) as tce:
+            ep = tm.get_endpoint_for("example.com", 1234)
+            self.assertIs(ep, exp_ep)
+            self.assertEqual(tce.mock_calls,
+                             [mock.call(b"example.com", 1234,
+                                        socks_endpoint=tse)])
+        with mock.patch("wormhole.tor_manager.TorClientEndpoint",
+                        return_value=exp_ep) as tce:
+            ep = tm.get_endpoint_for("127.0.0.1", 1234)
+            self.assertEqual(ep, None)
+            self.assertEqual(tce.mock_calls, [])

src/wormhole/tor_manager.py

@@ -1,37 +1,70 @@
 from __future__ import print_function, unicode_literals
-import time
+import sys, re
 from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.internet.error import ConnectError
-import txtorcon
+from twisted.internet.endpoints import clientFromString
+try:
+    from txtorcon import (TorConfig, launch_tor, build_tor_connection,
+                          DEFAULT_VALUE, TorClientEndpoint)
+except ImportError:
+    TorConfig = None
+    launch_tor = None
+    build_tor_connection = None
+    TorClientEndpoint = None
+    DEFAULT_VALUE = "DEFAULT_VALUE"
 import ipaddress
 from .timing import DebugTiming
 from .transit import allocate_tcp_port
 
 class TorManager:
-    def __init__(self, reactor, tor_socks_port=None, tor_control_port=9051,
+    def __init__(self, reactor, launch_tor=False, tor_control_port=None,
-                 timing=None):
+                 timing=None, stderr=sys.stderr):
         """
-        If tor_socks_port= is provided, I will assume that it points to a
+        If launch_tor=True, I will try to launch a new Tor process, ask it
-        functioning SOCKS server, and will use it for all outbound
+        for its SOCKS and control ports, and use those for outbound
-        connections. I will not attempt to establish a control-port
+        connections (and inbound onion-service listeners, if necessary).
-        connection, and I will not be able to run a server.
 
-        Otherwise, I will try to connect to an existing Tor process, first on
+        Otherwise if tor_control_port is provided, I will attempt to connect
-        localhost:9051, then /var/run/tor/control. Then I will try to
+        to an existing Tor's control port at the endpoint it specifies. I'll
-        authenticate, by reading a cookie file named by the Tor process. This
+        ask that Tor for its SOCKS port.
-        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.
+        With no arguments, I will try to connect to an existing Tor's control
+        port at the usual places: [unix:/var/run/tor/control,
+        tcp:127.0.0.1:9051, tcp:127.0.0.1:9151]. If any are successful, I'll
+        ask that Tor for its SOCKS port. If none are successful, I'll attempt
+        to do SOCKS to tcp:127.0.0.1:9050.
+
+        If I am unable to make a SOCKS connection, the initial connection to
+        the Rendezvous Server will fail, and the program will terminate.
+
+        Control-port connections can only succeed if I can authenticate (by
+        reading a cookie file named by the Tor process), so the current user
+        must have permission to read that file (either they started Tor, e.g.
+        TorBrowser, or they are in a unix group that's been given access,
+        e.g. debian-tor).
         """
+        # rationale: launching a new Tor takes a long time, so only do it if
+        # the user specifically asks for it with --launch-tor. Using an
+        # existing Tor should be much faster, but still requires general
+        # permission via --tor.
+
         self._reactor = reactor
-        # note: False is int
+        if not isinstance(launch_tor, bool): # note: False is int
-        assert isinstance(tor_socks_port, (int, type(None)))
+            raise TypeError("launch_tor= must be boolean")
-        assert isinstance(tor_control_port, int)
+        if not isinstance(tor_control_port, (type(""), type(None))):
-        self._tor_socks_port = tor_socks_port
+            raise TypeError("tor_control_port= must be str or None")
+        if launch_tor and tor_control_port is not None:
+            raise ValueError("cannot combine --launch-tor and --tor-control-port=")
+        self._launch_tor = launch_tor
         self._tor_control_port = tor_control_port
         self._timing = timing or DebugTiming()
+        self._stderr = stderr
+
+    def tor_available(self):
+        # unit tests mock out everything we get from txtorcon, so we can test
+        # this class under py3 even if txtorcon isn't installed. But the real
+        # commands need to know if they have Tor or not.
+        return bool(TorConfig)
 
     @inlineCallbacks
     def start(self):
@@ -40,73 +73,70 @@ class TorManager:
         # 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:
+        if self._launch_tor:
-            self._can_run_service = False
+            print(" launching a new Tor process, this may take a while..",
-            returnValue(True)
+                  file=self._stderr)
-
+            with self._timing.add("launch tor"):
-        _start_find = self._timing.add("find tor")
+                (tproto, tconfig, socks_desc) = yield self._do_launch_tor()
-        # try port 9051, then try /var/run/tor/control . Throws on failure.
-        state = None
-        with self._timing.add("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
-
-        if not state:
-            with self._timing.add("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
-
-        if state:
-            print("connected to pre-existing Tor process")
-            print("state:", state)
         else:
-            print("launching my own Tor process")
+            control_ports = ["unix:/var/run/tor/control", # debian tor package
-            yield self._create_my_own_tor()
+                             "tcp:127.0.0.1:9051", # standard Tor
-            # that sets self._tor_socks_port and self._tor_protocol
+                             "tcp:127.0.0.1:9151", # TorBrowser
+                             ]
+            if self._tor_control_port:
+                control_ports = [self._tor_control_port]
+            with self._timing.add("find tor"):
+                for control_port in control_ports:
+                    (tproto, tconfig,
+                     socks_desc) = yield self._try_control_port(control_port)
+                    if tproto:
+                        print(" using Tor (control port %s) (SOCKS port %s)"
+                              % (control_port, socks_desc),
+                              file=self._stderr)
+                        break
+                else:
+                    tproto = None
+                    tconfig = None
+                    socks_desc = "tcp:127.0.0.1:9050" # fallback
+                    print(" using Tor (SOCKS port %s)" % socks_desc,
+                          file=self._stderr)
 
-        _start_find.finish()
+        self._tor_protocol = tproto
-        self._can_run_service = True
+        self._tor_config = tconfig
-        returnValue(True)
+        self._tor_socks_endpoint = clientFromString(self._reactor, socks_desc)
 
     @inlineCallbacks
-    def _create_my_own_tor(self):
+    def _do_launch_tor(self):
-        with self._timing.add("launch tor"):
+        tconfig = TorConfig()
-            start = time.time()
+        #tconfig.ControlPort = allocate_tcp_port() # defaults to 9052
-            config = self.config = txtorcon.TorConfig()
+        tconfig.SocksPort = allocate_tcp_port()
-            if 0:
+        socks_desc = "tcp:127.0.0.1:%d" % tconfig.SocksPort
-                # The default is for launch_tor to create a tempdir itself,
+        # this could take tor_binary=
-                # and delete it when done. We only need to set a
+        tproto = yield launch_tor(tconfig, self._reactor)
-                # DataDirectory if we want it to be persistent.
+        returnValue((tproto, tconfig, socks_desc))
-                import tempfile
-                datadir = tempfile.mkdtemp()
-                config.DataDirectory = datadir
 
-            #config.ControlPort = allocate_tcp_port() # defaults to 9052
+    @inlineCallbacks
-            #print("setting config.ControlPort to", config.ControlPort)
+    def _try_control_port(self, control_port):
-            config.SocksPort = allocate_tcp_port()
+        NOPE = (None, None, None)
-            self._tor_socks_port = config.SocksPort
+        ep = clientFromString(self._reactor, control_port)
-            print("setting config.SocksPort to", config.SocksPort)
+        try:
-
+            tproto = yield build_tor_connection(ep, build_state=False)
-            tpp = yield txtorcon.launch_tor(config, self._reactor,
+            # now wait for bootstrap
-                                            #tor_binary=
+            tconfig = yield TorConfig.from_protocol(tproto)
-                                            )
+        except (ValueError, ConnectError):
-            # gives a TorProcessProtocol with .tor_protocol
+            returnValue(NOPE)
-            self._tor_protocol = tpp.tor_protocol
+        socks_ports = list(tconfig.SocksPort)
-            print("tp:", self._tor_protocol)
+        socks_port = socks_ports[0] # TODO: when might there be multiple?
-            print("elapsed:", time.time() - start)
+        # I've seen "9050", and "unix:/var/run/tor/socks WorldWritable"
-        returnValue(True)
+        pieces = socks_port.split()
+        p = pieces[0]
+        if p == DEFAULT_VALUE:
+            socks_desc = "tcp:127.0.0.1:9050"
+        elif re.search('^\d+$', p):
+            socks_desc = "tcp:127.0.0.1:%s" % p
+        else:
+            socks_desc = p
+        returnValue((tproto, tconfig, socks_desc))
 
     def is_non_public_numeric_address(self, host):
         # for numeric hostnames, skip RFC1918 addresses, since no Tor exit
@@ -125,14 +155,12 @@ class TorManager:
     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,
+        ep = TorClientEndpoint(host, port,
-                                        socks_hostname="127.0.0.1",
+                               socks_endpoint=self._tor_socks_endpoint)
-                                        socks_port=self._tor_socks_port)
         return ep

src/wormhole/xfer_util.py

@@ -2,10 +2,13 @@ import json
 from twisted.internet.defer import inlineCallbacks, returnValue
 
 from .wormhole import wormhole
-
+from .tor_manager import TorManager
+from .errors import NoTorError
 
 @inlineCallbacks
-def receive(reactor, appid, relay_url, code, use_tor=None, on_code=None):
+def receive(reactor, appid, relay_url, code,
+            use_tor=False, launch_tor=False, tor_control_port=None,
+            on_code=None):
     """
     This is a convenience API which returns a Deferred that callbacks
     with a single chunk of data from another wormhole (and then closes
@@ -24,7 +27,18 @@ def receive(reactor, appid, relay_url, code, use_tor=None, on_code=None):
     :param on_code: if not None, this is called when we have a code (even if you passed in one explicitly)
     :type on_code: single-argument callable
     """
-    wh = wormhole(appid, relay_url, reactor, use_tor)
+    tm = None
+    if use_tor:
+        tm = TorManager(reactor, launch_tor, tor_control_port)
+        # 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
+        if not tm.tor_available():
+            raise NoTorError()
+        yield tm.start()
+
+    wh = wormhole(appid, relay_url, reactor, tor_manager=tm)
     if code is None:
         code = yield wh.get_code()
     else:
@@ -55,7 +69,9 @@ def receive(reactor, appid, relay_url, code, use_tor=None, on_code=None):
 
 
 @inlineCallbacks
-def send(reactor, appid, relay_url, data, code, use_tor=None, on_code=None):
+def send(reactor, appid, relay_url, data, code,
+         use_tor=False, launch_tor=False, tor_control_port=None,
+         on_code=None):
     """
     This is a convenience API which returns a Deferred that callbacks
     after a single chunk of data has been sent to another
@@ -74,7 +90,17 @@ def send(reactor, appid, relay_url, data, code, use_tor=None, on_code=None):
     :param on_code: if not None, this is called when we have a code (even if you passed in one explicitly)
     :type on_code: single-argument callable
     """
-    wh = wormhole(appid, relay_url, reactor, use_tor)
+    tm = None
+    if use_tor:
+        tm = TorManager(reactor, launch_tor, tor_control_port)
+        # 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
+        if not tm.tor_available():
+            raise NoTorError()
+        yield tm.start()
+    wh = wormhole(appid, relay_url, reactor, tor_manager=tm)
     if code is None:
         code = yield wh.get_code()
     else: