Implement i2p_provider and --listen=i2p

Closes ticket:2838

setup.py

@@ -275,7 +275,7 @@ setup(name="tahoe-lafs", # also set in __init__.py
               "foolscap[tor] >= 0.12.3",
               "txtorcon >= 0.17.0", # in case pip's resolver doesn't work
               "foolscap[i2p]",
-              "txi2p", # in case pip's resolver doesn't work
+              "txi2p >= 0.3.1", # in case pip's resolver doesn't work
               "pytest",
               "pytest-twisted",
           ],
@@ -285,7 +285,7 @@ setup(name="tahoe-lafs", # also set in __init__.py
           ],
           "i2p": [
               "foolscap[i2p]",
-              "txi2p", # in case pip's resolver doesn't work
+              "txi2p >= 0.3.1", # in case pip's resolver doesn't work
           ],
       },
       package_data={"allmydata.web": ["*.xhtml",

src/allmydata/node.py

@@ -1,7 +1,7 @@
 import datetime, os.path, re, types, ConfigParser, tempfile
 from base64 import b32decode, b32encode
 
-from twisted.internet import reactor, endpoints
+from twisted.internet import reactor
 from twisted.python import log as twlog
 from twisted.application import service
 from foolscap.api import Tub, app_versions
@@ -13,14 +13,7 @@ from allmydata.util.assertutil import _assert
 from allmydata.util.fileutil import abspath_expanduser_unicode
 from allmydata.util.encodingutil import get_filesystem_encoding, quote_output
 from allmydata.util import configutil
-from allmydata.util import tor_provider
-
-def _import_i2p():
-    try:
-        from foolscap.connections import i2p
-        return i2p
-    except ImportError: # pragma: no cover
-        return None
+from allmydata.util import i2p_provider, tor_provider
 
 def _common_config_sections():
     return {
@@ -45,6 +38,9 @@ def _common_config_sections():
             "i2p.executable",
             "launch",
             "sam.port",
+            "dest",
+            "dest.port",
+            "dest.private_key_file",
         ),
         "tor": (
             "control.port",
@@ -140,6 +136,7 @@ class Node(service.MultiService):
         self.logSource="Node"
         self.setup_logging()
 
+        self.create_i2p_provider()
         self.create_tor_provider()
         self.init_connections()
         self.set_tub_options()
@@ -222,6 +219,11 @@ class Node(service.MultiService):
     def check_privacy(self):
         self._reveal_ip = self.get_config("node", "reveal-IP-address", True,
                                           boolean=True)
+    def create_i2p_provider(self):
+        self._i2p_provider = i2p_provider.Provider(self.basedir, self, reactor)
+        self._i2p_provider.check_dest_config()
+        self._i2p_provider.setServiceParent(self)
+
     def create_tor_provider(self):
         self._tor_provider = tor_provider.Provider(self.basedir, self, reactor)
         self._tor_provider.check_onion_config()
@@ -236,32 +238,7 @@ class Node(service.MultiService):
         return self._tor_provider.get_tor_handler()
 
     def _make_i2p_handler(self):
-        enabled = self.get_config("i2p", "enabled", True, boolean=True)
-        if not enabled:
-            return None
-        i2p = _import_i2p()
-        if not i2p:
-            return None
-
-        samport = self.get_config("i2p", "sam.port", None)
-        launch = self.get_config("i2p", "launch", False, boolean=True)
-        configdir = self.get_config("i2p", "i2p.configdir", None)
-
-        if samport:
-            if launch:
-                raise ValueError("tahoe.cfg [i2p] must not set both "
-                                 "sam.port and launch")
-            ep = endpoints.clientFromString(reactor, samport)
-            return i2p.sam_endpoint(ep)
-
-        if launch:
-            executable = self.get_config("i2p", "i2p.executable", None)
-            return i2p.launch(i2p_configdir=configdir, i2p_binary=executable)
-
-        if configdir:
-            return i2p.local_i2p(configdir)
-
-        return i2p.default(reactor)
+        return self._i2p_provider.get_i2p_handler()
 
     def init_connections(self):
         # We store handlers for everything. None means we were unable to

src/allmydata/scripts/create_node.py

@@ -5,7 +5,7 @@ from allmydata.scripts.common import BasedirOptions, NoDefaultBasedirOptions
 from allmydata.scripts.default_nodedir import _default_nodedir
 from allmydata.util.assertutil import precondition
 from allmydata.util.encodingutil import listdir_unicode, argv_to_unicode, quote_local_unicode_path
-from allmydata.util import fileutil, iputil, tor_provider
+from allmydata.util import fileutil, i2p_provider, iputil, tor_provider
 
 
 dummy_tac = """
@@ -41,6 +41,17 @@ TOR_FLAGS = [
     ("tor-launch", None, "Launch a tor instead of connecting to a tor control port."),
 ]
 
+I2P_OPTS = [
+    ("i2p-sam-port", None, None,
+     "I2P's SAM API port endpoint descriptor string (e.g. tcp:127.0.0.1:7656)"),
+    ("i2p-executable", None, None,
+     "(future) The 'i2prouter' executable to run (default is to search $PATH)."),
+]
+
+I2P_FLAGS = [
+    ("i2p-launch", None, "(future) Launch an I2P router instead of connecting to a SAM API port."),
+]
+
 def validate_where_options(o):
     if o['listen'] == "none":
         # no other arguments are accepted
@@ -89,6 +100,18 @@ def validate_tor_options(o):
     if o["tor-launch"] and o["tor-control-port"]:
         raise UsageError("use either --tor-launch or --tor-control-port=, not both")
 
+def validate_i2p_options(o):
+    use_i2p = "i2p" in o["listen"].split(",")
+    if not use_i2p:
+        if o["i2p-launch"]:
+            raise UsageError("--i2p-launch requires --listen=i2p")
+        if o["i2p-sam-port"]:
+            raise UsageError("--i2p-sam-port= requires --listen=i2p")
+    if o["i2p-launch"] and o["i2p-sam-port"]:
+        raise UsageError("use either --i2p-launch or --i2p-sam-port=, not both")
+    if o["i2p-launch"]:
+        raise UsageError("--i2p-launch is under development")
+
 class _CreateBaseOptions(BasedirOptions):
     optFlags = [
         ("hide-ip", None, "prohibit any configuration that would reveal the node's IP address"),
@@ -119,28 +142,30 @@ class CreateClientOptions(_CreateBaseOptions):
 class CreateNodeOptions(CreateClientOptions):
     optFlags = [
         ("no-storage", None, "Do not offer storage service to other nodes."),
-        ] + TOR_FLAGS
+        ] + TOR_FLAGS + I2P_FLAGS
 
     synopsis = "[options] [NODEDIR]"
     description = "Create a full Tahoe-LAFS node (client+server)."
-    optParameters = CreateClientOptions.optParameters + WHERE_OPTS + TOR_OPTS
+    optParameters = CreateClientOptions.optParameters + WHERE_OPTS + TOR_OPTS + I2P_OPTS
 
     def parseArgs(self, basedir=None):
         CreateClientOptions.parseArgs(self, basedir)
         validate_where_options(self)
         validate_tor_options(self)
+        validate_i2p_options(self)
 
 class CreateIntroducerOptions(NoDefaultBasedirOptions):
     subcommand_name = "create-introducer"
     description = "Create a Tahoe-LAFS introducer."
     optFlags = [
         ("hide-ip", None, "prohibit any configuration that would reveal the node's IP address"),
-    ] + TOR_FLAGS
-    optParameters = NoDefaultBasedirOptions.optParameters + WHERE_OPTS + TOR_OPTS
+    ] + TOR_FLAGS + I2P_FLAGS
+    optParameters = NoDefaultBasedirOptions.optParameters + WHERE_OPTS + TOR_OPTS + I2P_OPTS
     def parseArgs(self, basedir=None):
         NoDefaultBasedirOptions.parseArgs(self, basedir)
         validate_where_options(self)
         validate_tor_options(self)
+        validate_i2p_options(self)
 
 @defer.inlineCallbacks
 def write_node_config(c, config):
@@ -177,6 +202,7 @@ def write_node_config(c, config):
     listeners = config['listen'].split(",")
 
     tor_config = {}
+    i2p_config = {}
     tub_ports = []
     tub_locations = []
     if listeners == ["none"]:
@@ -189,8 +215,10 @@ def write_node_config(c, config):
             tub_ports.append(tor_port)
             tub_locations.append(tor_location)
         if "i2p" in listeners:
-            raise NotImplementedError("--listen=i2p is under development, "
-                                      "see ticket #2490 for details")
+            (i2p_config, i2p_port, i2p_location) = \
+                         yield i2p_provider.create_dest(reactor, config)
+            tub_ports.append(i2p_port)
+            tub_locations.append(i2p_location)
         if "tcp" in listeners:
             if config["port"]: # --port/--location are a pair
                 tub_ports.append(config["port"].encode('utf-8'))
@@ -219,6 +247,12 @@ def write_node_config(c, config):
             c.write("%s = %s\n" % (key, value))
         c.write("\n")
 
+    if i2p_config:
+        c.write("[i2p]\n")
+        for key, value in i2p_config.items():
+            c.write("%s = %s\n" % (key, value))
+        c.write("\n")
+
 
 def write_client_config(c, config):
     c.write("[client]\n")

src/allmydata/test/cli/test_create.py

@@ -156,14 +156,6 @@ class Config(unittest.TestCase):
                               basedir)
         self.assertEqual(str(e), "--listen= must be none, or one/some of: tcp, tor, i2p")
 
-    @defer.inlineCallbacks
-    def test_node_listen_i2p(self):
-        basedir = self.mktemp()
-        d = run_cli("create-node", "--listen=i2p", basedir)
-        e = yield self.assertFailure(d, NotImplementedError)
-        self.assertEqual(str(e), "--listen=i2p is under development, "
-                         "see ticket #2490 for details")
-
     def test_node_listen_tor_hostname(self):
         e = self.assertRaises(usage.UsageError,
                               parse_cli,
@@ -208,6 +200,19 @@ class Config(unittest.TestCase):
         self.assertIn("Node created", out)
         self.assertEqual(err, "")
 
+    @defer.inlineCallbacks
+    def test_node_slow_i2p(self):
+        basedir = self.mktemp()
+        d = defer.Deferred()
+        with mock.patch("allmydata.util.i2p_provider.create_dest",
+                        return_value=d):
+            d2 = run_cli("create-node", "--listen=i2p", basedir)
+            d.callback(({}, "port", "location"))
+            rc, out, err = yield d2
+        self.assertEqual(rc, 0)
+        self.assertIn("Node created", out)
+        self.assertEqual(err, "")
+
     def test_introducer_no_hostname(self):
         basedir = self.mktemp()
         e = self.assertRaises(usage.UsageError, parse_cli,
@@ -317,3 +322,68 @@ class Tor(unittest.TestCase):
                               "create-node", "--listen=none",
                               "--tor-control-port=foo")
         self.assertEqual(str(e), "--tor-control-port= requires --listen=tor")
+
+class I2P(unittest.TestCase):
+    def test_default(self):
+        basedir = self.mktemp()
+        i2p_config = {"abc": "def"}
+        i2p_port = "ghi"
+        i2p_location = "jkl"
+        dest_d = defer.succeed( (i2p_config, i2p_port, i2p_location) )
+        with mock.patch("allmydata.util.i2p_provider.create_dest",
+                        return_value=dest_d) as co:
+            rc, out, err = self.successResultOf(
+                run_cli("create-node", "--listen=i2p", basedir))
+        self.assertEqual(len(co.mock_calls), 1)
+        args = co.mock_calls[0][1]
+        self.assertIdentical(args[0], reactor)
+        self.assertIsInstance(args[1], create_node.CreateNodeOptions)
+        self.assertEqual(args[1]["listen"], "i2p")
+        cfg = read_config(basedir)
+        self.assertEqual(cfg.get("i2p", "abc"), "def")
+        self.assertEqual(cfg.get("node", "tub.port"), "ghi")
+        self.assertEqual(cfg.get("node", "tub.location"), "jkl")
+
+    def test_launch(self):
+        e = self.assertRaises(usage.UsageError,
+                              parse_cli,
+                              "create-node", "--listen=i2p", "--i2p-launch")
+        self.assertEqual(str(e), "--i2p-launch is under development")
+
+
+    def test_sam_port(self):
+        basedir = self.mktemp()
+        i2p_config = {"abc": "def"}
+        i2p_port = "ghi"
+        i2p_location = "jkl"
+        dest_d = defer.succeed( (i2p_config, i2p_port, i2p_location) )
+        with mock.patch("allmydata.util.i2p_provider.create_dest",
+                        return_value=dest_d) as co:
+            rc, out, err = self.successResultOf(
+                run_cli("create-node", "--listen=i2p", "--i2p-sam-port=mno",
+                        basedir))
+        args = co.mock_calls[0][1]
+        self.assertEqual(args[1]["listen"], "i2p")
+        self.assertEqual(args[1]["i2p-launch"], False)
+        self.assertEqual(args[1]["i2p-sam-port"], "mno")
+
+    def test_not_both(self):
+        e = self.assertRaises(usage.UsageError,
+                              parse_cli,
+                              "create-node", "--listen=i2p",
+                              "--i2p-launch", "--i2p-sam-port=foo")
+        self.assertEqual(str(e), "use either --i2p-launch or"
+                         " --i2p-sam-port=, not both")
+
+    def test_launch_without_listen(self):
+        e = self.assertRaises(usage.UsageError,
+                              parse_cli,
+                              "create-node", "--listen=none", "--i2p-launch")
+        self.assertEqual(str(e), "--i2p-launch requires --listen=i2p")
+
+    def test_sam_port_without_listen(self):
+        e = self.assertRaises(usage.UsageError,
+                              parse_cli,
+                              "create-node", "--listen=none",
+                              "--i2p-sam-port=foo")
+        self.assertEqual(str(e), "--i2p-sam-port= requires --listen=i2p")

src/allmydata/test/test_connections.py

@@ -15,6 +15,7 @@ class FakeNode(Node):
         self._reveal_ip = True
         self.basedir = "BASEDIR"
         self.services = []
+        self.create_i2p_provider()
         self.create_tor_provider()
 
 BASECONFIG = ("[client]\n"
@@ -141,8 +142,9 @@ class I2P(unittest.TestCase):
         self.assertEqual(h, None)
 
     def test_unimportable(self):
-        n = FakeNode(BASECONFIG)
-        with mock.patch("allmydata.node._import_i2p", return_value=None):
+        with mock.patch("allmydata.util.i2p_provider._import_i2p",
+                        return_value=None):
+            n = FakeNode(BASECONFIG)
             h = n._make_i2p_handler()
         self.assertEqual(h, None)
 

src/allmydata/test/test_i2p_provider.py

@@ -0,0 +1,356 @@
+import os
+from twisted.trial import unittest
+from twisted.internet import defer, error
+from twisted.python.usage import UsageError
+from StringIO import StringIO
+import mock
+from ..util import i2p_provider
+from ..scripts import create_node, runner
+
+def mock_txi2p(txi2p):
+    return mock.patch("allmydata.util.i2p_provider._import_txi2p",
+                      return_value=txi2p)
+
+def mock_i2p(i2p):
+    return mock.patch("allmydata.util.i2p_provider._import_i2p",
+                      return_value=i2p)
+
+def make_cli_config(basedir, *argv):
+    parent = runner.Options()
+    cli_config = create_node.CreateNodeOptions()
+    cli_config.parent = parent
+    cli_config.parseOptions(argv)
+    cli_config["basedir"] = basedir
+    cli_config.stdout = StringIO()
+    return cli_config
+
+class TryToConnect(unittest.TestCase):
+    def test_try(self):
+        reactor = object()
+        txi2p = mock.Mock()
+        d = defer.succeed(True)
+        txi2p.testAPI = mock.Mock(return_value=d)
+        ep = object()
+        stdout = StringIO()
+        with mock.patch("allmydata.util.i2p_provider.clientFromString",
+                        return_value=ep) as cfs:
+            d = i2p_provider._try_to_connect(reactor, "desc", stdout, txi2p)
+        r = self.successResultOf(d)
+        self.assertTrue(r)
+        cfs.assert_called_with(reactor, "desc")
+        txi2p.testAPI.assert_called_with(reactor, 'SAM', ep)
+
+    def test_try_handled_error(self):
+        reactor = object()
+        txi2p = mock.Mock()
+        d = defer.fail(error.ConnectError("oops"))
+        txi2p.testAPI = mock.Mock(return_value=d)
+        ep = object()
+        stdout = StringIO()
+        with mock.patch("allmydata.util.i2p_provider.clientFromString",
+                        return_value=ep) as cfs:
+            d = i2p_provider._try_to_connect(reactor, "desc", stdout, txi2p)
+        r = self.successResultOf(d)
+        self.assertIs(r, None)
+        cfs.assert_called_with(reactor, "desc")
+        txi2p.testAPI.assert_called_with(reactor, 'SAM', ep)
+        self.assertEqual(stdout.getvalue(),
+                         "Unable to reach I2P SAM API at 'desc': "
+                         "An error occurred while connecting: oops.\n")
+
+    def test_try_unhandled_error(self):
+        reactor = object()
+        txi2p = mock.Mock()
+        d = defer.fail(ValueError("oops"))
+        txi2p.testAPI = mock.Mock(return_value=d)
+        ep = object()
+        stdout = StringIO()
+        with mock.patch("allmydata.util.i2p_provider.clientFromString",
+                        return_value=ep) as cfs:
+            d = i2p_provider._try_to_connect(reactor, "desc", stdout, txi2p)
+        f = self.failureResultOf(d)
+        self.assertIsInstance(f.value, ValueError)
+        self.assertEqual(str(f.value), "oops")
+        cfs.assert_called_with(reactor, "desc")
+        txi2p.testAPI.assert_called_with(reactor, 'SAM', ep)
+        self.assertEqual(stdout.getvalue(), "")
+
+class ConnectToI2P(unittest.TestCase):
+    def _do_test_connect(self, endpoint, reachable):
+        reactor = object()
+        txi2p = object()
+        args = []
+        if endpoint:
+            args = ["--i2p-sam-port=%s" % endpoint]
+        cli_config = make_cli_config("basedir", "--listen=i2p", *args)
+        stdout = cli_config.stdout
+        expected_port = "tcp:127.0.0.1:7656"
+        if endpoint:
+            expected_port = endpoint
+        tried = []
+        def _try_to_connect(reactor, port, stdout, txi2p):
+            tried.append( (reactor, port, stdout, txi2p) )
+            if not reachable:
+                return defer.succeed(None)
+            if port == expected_port:
+                return defer.succeed(True)
+            return defer.succeed(None)
+
+        with mock.patch("allmydata.util.i2p_provider._try_to_connect",
+                        _try_to_connect):
+            d = i2p_provider._connect_to_i2p(reactor, cli_config, txi2p)
+        if not reachable:
+            f = self.failureResultOf(d)
+            self.assertIsInstance(f.value, ValueError)
+            self.assertEqual(str(f.value),
+                             "unable to reach any default I2P SAM port")
+            return
+        successful_port = self.successResultOf(d)
+        self.assertEqual(successful_port, expected_port)
+        expected = [(reactor, "tcp:127.0.0.1:7656", stdout, txi2p)]
+        if endpoint:
+            expected = [(reactor, endpoint, stdout, txi2p)]
+        self.assertEqual(tried, expected)
+
+    def test_connect(self):
+        return self._do_test_connect(None, True)
+    def test_connect_endpoint(self):
+        return self._do_test_connect("tcp:other:port", True)
+    def test_connect_unreachable(self):
+        return self._do_test_connect(None, False)
+
+
+class CreateDest(unittest.TestCase):
+    def test_no_txi2p(self):
+        with mock.patch("allmydata.util.i2p_provider._import_txi2p",
+                        return_value=None):
+            d = i2p_provider.create_dest("reactor", "cli_config")
+            f = self.failureResultOf(d)
+            self.assertIsInstance(f.value, ValueError)
+            self.assertEqual(str(f.value),
+                             "Cannot create I2P Destination without txi2p. "
+                             "Please 'pip install tahoe-lafs[i2p]' to fix this.")
+
+    def _do_test_launch(self, executable):
+        basedir = self.mktemp()
+        os.mkdir(basedir)
+        args = ["--listen=i2p", "--i2p-launch"]
+        if executable:
+            args.append("--i2p-executable=%s" % executable)
+        self.assertRaises(UsageError, make_cli_config, basedir, *args)
+
+    def test_launch(self):
+        return self._do_test_launch(None)
+    def test_launch_executable(self):
+        return self._do_test_launch("myi2p")
+
+    def test_sam_endpoint(self):
+        basedir = self.mktemp()
+        os.mkdir(basedir)
+        private_dir = os.path.join(basedir, "private")
+        os.mkdir(private_dir)
+        privkeyfile = os.path.abspath(os.path.join(private_dir, "i2p_dest.privkey"))
+        reactor = object()
+        cli_config = make_cli_config(basedir, "--listen=i2p")
+        connect_to_i2p = mock.Mock(return_value=defer.succeed("goodport"))
+        txi2p = mock.Mock()
+        ep = object()
+        dest = mock.Mock()
+        dest.host = "FOOBAR.b32.i2p"
+        txi2p.generateDestination = mock.Mock(return_value=defer.succeed(dest))
+
+        with mock_txi2p(txi2p):
+            with mock.patch("allmydata.util.i2p_provider._connect_to_i2p",
+                            connect_to_i2p):
+                with mock.patch("allmydata.util.i2p_provider.clientFromString",
+                                return_value=ep) as cfs:
+                    d = i2p_provider.create_dest(reactor, cli_config)
+        tahoe_config_i2p, i2p_port, i2p_location = self.successResultOf(d)
+
+        connect_to_i2p.assert_called_with(reactor, cli_config, txi2p)
+        cfs.assert_called_with(reactor, "goodport")
+        txi2p.generateDestination.assert_called_with(reactor, privkeyfile, 'SAM', ep)
+
+        expected = {"sam.port": "goodport",
+                    "dest": "true",
+                    "dest.port": "3457",
+                    "dest.private_key_file": os.path.join("private",
+                                                          "i2p_dest.privkey"),
+                    }
+        self.assertEqual(tahoe_config_i2p, expected)
+        self.assertEqual(i2p_port, "i2p:%s:3457:api=SAM:apiEndpoint=goodport" % privkeyfile)
+        self.assertEqual(i2p_location, "i2p:FOOBAR.b32.i2p:3457")
+
+_None = object()
+class FakeConfig(dict):
+    def get_config(self, section, option, default=_None, boolean=False):
+        if section != "i2p":
+            raise ValueError(section)
+        value = self.get(option, default)
+        if value is _None:
+            raise KeyError
+        return value
+
+class Provider(unittest.TestCase):
+    def test_build(self):
+        i2p_provider.Provider("basedir", FakeConfig(), "reactor")
+
+    def test_handler_disabled(self):
+        p = i2p_provider.Provider("basedir", FakeConfig(enabled=False),
+                                  "reactor")
+        self.assertEqual(p.get_i2p_handler(), None)
+
+    def test_handler_no_i2p(self):
+        with mock_i2p(None):
+            p = i2p_provider.Provider("basedir", FakeConfig(), "reactor")
+        self.assertEqual(p.get_i2p_handler(), None)
+
+    def test_handler_sam_endpoint(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.sam_endpoint = mock.Mock(return_value=handler)
+        ep = object()
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir",
+                                      FakeConfig(**{"sam.port": "ep_desc"}),
+                                      reactor)
+            with mock.patch("allmydata.util.i2p_provider.clientFromString",
+                            return_value=ep) as cfs:
+                h = p.get_i2p_handler()
+        cfs.assert_called_with(reactor, "ep_desc")
+        self.assertIs(h, handler)
+        i2p.sam_endpoint.assert_called_with(ep)
+
+    def test_handler_launch(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.launch = mock.Mock(return_value=handler)
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir", FakeConfig(launch=True),
+                                      reactor)
+        h = p.get_i2p_handler()
+        self.assertIs(h, handler)
+        i2p.launch.assert_called_with(i2p_configdir=None, i2p_binary=None)
+
+    def test_handler_launch_configdir(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.launch = mock.Mock(return_value=handler)
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir",
+                                      FakeConfig(launch=True,
+                                                 **{"i2p.configdir": "configdir"}),
+                                      reactor)
+        h = p.get_i2p_handler()
+        self.assertIs(h, handler)
+        i2p.launch.assert_called_with(i2p_configdir="configdir", i2p_binary=None)
+
+    def test_handler_launch_configdir_executable(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.launch = mock.Mock(return_value=handler)
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir",
+                                      FakeConfig(launch=True,
+                                                 **{"i2p.configdir": "configdir",
+                                                    "i2p.executable": "myi2p",
+                                                   }),
+                                      reactor)
+        h = p.get_i2p_handler()
+        self.assertIs(h, handler)
+        i2p.launch.assert_called_with(i2p_configdir="configdir", i2p_binary="myi2p")
+
+    def test_handler_configdir(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.local_i2p = mock.Mock(return_value=handler)
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir",
+                                      FakeConfig(**{"i2p.configdir": "configdir"}),
+                                      reactor)
+        h = p.get_i2p_handler()
+        i2p.local_i2p.assert_called_with("configdir")
+        self.assertIs(h, handler)
+
+    def test_handler_default(self):
+        i2p = mock.Mock()
+        handler = object()
+        i2p.default = mock.Mock(return_value=handler)
+        reactor = object()
+
+        with mock_i2p(i2p):
+            p = i2p_provider.Provider("basedir", FakeConfig(), reactor)
+        h = p.get_i2p_handler()
+        self.assertIs(h, handler)
+        i2p.default.assert_called_with(reactor)
+
+class Provider_CheckI2PConfig(unittest.TestCase):
+    def test_default(self):
+        # default config doesn't start an I2P service, so it should be
+        # happy both with and without txi2p
+
+        p = i2p_provider.Provider("basedir", FakeConfig(), "reactor")
+        p.check_dest_config()
+
+        with mock_txi2p(None):
+            p = i2p_provider.Provider("basedir", FakeConfig(), "reactor")
+            p.check_dest_config()
+
+    def test_no_txi2p(self):
+        with mock_txi2p(None):
+            p = i2p_provider.Provider("basedir", FakeConfig(dest=True),
+                                      "reactor")
+            e = self.assertRaises(ValueError, p.check_dest_config)
+            self.assertEqual(str(e), "Cannot create I2P Destination without txi2p. "
+                             "Please 'pip install tahoe-lafs[i2p]' to fix.")
+
+    def test_no_launch_no_control(self):
+        p = i2p_provider.Provider("basedir", FakeConfig(dest=True), "reactor")
+        e = self.assertRaises(ValueError, p.check_dest_config)
+        self.assertEqual(str(e), "[i2p] dest = true, but we have neither "
+                         "sam.port= nor launch=true nor configdir=")
+
+    def test_missing_keys(self):
+        p = i2p_provider.Provider("basedir", FakeConfig(dest=True,
+                                             **{"sam.port": "x",
+                                                }), "reactor")
+        e = self.assertRaises(ValueError, p.check_dest_config)
+        self.assertEqual(str(e), "[i2p] dest = true, "
+                         "but dest.port= is missing")
+
+        p = i2p_provider.Provider("basedir",
+                                  FakeConfig(dest=True,
+                                             **{"sam.port": "x",
+                                                "dest.port": "y",
+                                                }), "reactor")
+        e = self.assertRaises(ValueError, p.check_dest_config)
+        self.assertEqual(str(e), "[i2p] dest = true, "
+                         "but dest.private_key_file= is missing")
+
+    def test_launch_not_implemented(self):
+        p = i2p_provider.Provider("basedir",
+                                  FakeConfig(dest=True, launch=True,
+                                             **{"dest.port": "x",
+                                                "dest.private_key_file": "y",
+                                                }), "reactor")
+        e = self.assertRaises(NotImplementedError, p.check_dest_config)
+        self.assertEqual(str(e), "[i2p] launch is under development.")
+
+    def test_ok(self):
+        p = i2p_provider.Provider("basedir",
+                                  FakeConfig(dest=True,
+                                             **{"sam.port": "x",
+                                                "dest.port": "y",
+                                                "dest.private_key_file": "z",
+                                                }), "reactor")
+        p.check_dest_config()

src/allmydata/test/test_node.py

@@ -355,6 +355,7 @@ class MultiplePorts(unittest.TestCase):
         n.read_config()
         n.check_privacy()
         n.services = []
+        n.create_i2p_provider()
         n.create_tor_provider()
         n.init_connections()
         n.set_tub_options()

src/allmydata/util/i2p_provider.py

@@ -0,0 +1,202 @@
+# -*- coding: utf-8 -*-
+from __future__ import absolute_import, print_function, with_statement
+import os
+
+from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.endpoints import clientFromString
+from twisted.internet.error import ConnectionRefusedError, ConnectError
+from twisted.application import service
+
+def _import_i2p():
+    # this exists to be overridden by unit tests
+    try:
+        from foolscap.connections import i2p
+        return i2p
+    except ImportError: # pragma: no cover
+        return None
+
+def _import_txi2p():
+    try:
+        import txi2p
+        return txi2p
+    except ImportError: # pragma: no cover
+        return None
+
+
+def _try_to_connect(reactor, endpoint_desc, stdout, txi2p):
+    # yields True or None
+    ep = clientFromString(reactor, endpoint_desc)
+    d = txi2p.testAPI(reactor, 'SAM', ep)
+    def _failed(f):
+        # depending upon what's listening at that endpoint, we might get
+        # various errors. If this list is too short, we might expose an
+        # exception to the user (causing "tahoe create-node" to fail messily)
+        # when we're supposed to just try the next potential port instead.
+        # But I don't want to catch everything, because that may hide actual
+        # coding errors.
+        f.trap(ConnectionRefusedError, # nothing listening on TCP
+               ConnectError, # missing unix socket, or permission denied
+               #ValueError,
+               # connecting to e.g. an HTTP server causes an
+               # UnhandledException (around a ValueError) when the handshake
+               # fails to parse, but that's not something we can catch. The
+               # attempt hangs, so don't do that.
+               RuntimeError, # authentication failure
+               )
+        if stdout:
+            stdout.write("Unable to reach I2P SAM API at '%s': %s\n" %
+                         (endpoint_desc, f.value))
+        return None
+    d.addErrback(_failed)
+    return d
+
+@inlineCallbacks
+def _connect_to_i2p(reactor, cli_config, txi2p):
+    # we assume i2p is already running
+    ports_to_try = ["tcp:127.0.0.1:7656"]
+    if cli_config["i2p-sam-port"]:
+        ports_to_try = [cli_config["i2p-sam-port"]]
+    for port in ports_to_try:
+        accessible = yield _try_to_connect(reactor, port, cli_config.stdout,
+                                           txi2p)
+        if accessible:
+            returnValue(port) ; break # helps editor
+    else:
+        raise ValueError("unable to reach any default I2P SAM port")
+
+@inlineCallbacks
+def create_dest(reactor, cli_config):
+    txi2p = _import_txi2p()
+    if not txi2p:
+        raise ValueError("Cannot create I2P Destination without txi2p. "
+                         "Please 'pip install tahoe-lafs[i2p]' to fix this.")
+    tahoe_config_i2p = {} # written into tahoe.cfg:[i2p]
+    private_dir = os.path.abspath(os.path.join(cli_config["basedir"], "private"))
+    stdout = cli_config.stdout
+    if cli_config["i2p-launch"]:
+        raise NotImplementedError("--i2p-launch is under development.")
+    else:
+        print("connecting to I2P (to allocate .i2p address)..", file=stdout)
+        sam_port = yield _connect_to_i2p(reactor, cli_config, txi2p)
+        print("I2P connection established", file=stdout)
+        tahoe_config_i2p["sam.port"] = sam_port
+
+    external_port = 3457 # TODO: pick this randomly? there's no contention.
+
+    privkeyfile = os.path.join(private_dir, "i2p_dest.privkey")
+    sam_endpoint = clientFromString(reactor, sam_port)
+    print("allocating .i2p address...", file=stdout)
+    dest = yield txi2p.generateDestination(reactor, privkeyfile, 'SAM', sam_endpoint)
+    print(".i2p address allocated", file=stdout)
+    escaped_sam_port = sam_port.replace(':', '\:')
+    i2p_port = "i2p:%s:%d:api=SAM:apiEndpoint=%s" % \
+        (privkeyfile, external_port, escaped_sam_port)
+    i2p_location = "i2p:%s:%d" % (dest.host, external_port)
+
+    # in addition to the "how to launch/connect-to i2p" keys above, we also
+    # record information about the I2P service into tahoe.cfg.
+    # * "port" is the random "public Destination port" (integer), which
+    #   (when combined with the .i2p address) should match "i2p_location"
+    #   (which will be added to tub.location)
+    # * "private_key_file" points to the on-disk copy of the private key
+    #   material (although we always write it to the same place)
+
+    tahoe_config_i2p["dest"] = "true"
+    tahoe_config_i2p["dest.port"] = str(external_port)
+    tahoe_config_i2p["dest.private_key_file"] = os.path.join("private",
+                                                             "i2p_dest.privkey")
+
+    # tahoe_config_i2p: this is a dictionary of keys/values to add to the
+    # "[i2p]" section of tahoe.cfg, which tells the new node how to launch
+    # I2P in the right way.
+
+    # i2p_port: a server endpoint string, it will be added to tub.port=
+
+    # i2p_location: a foolscap connection hint, "i2p:B32_ADDR:PORT"
+
+    # We assume/require that the Node gives us the same data_directory=
+    # at both create-node and startup time. The data directory is not
+    # recorded in tahoe.cfg
+
+    returnValue((tahoe_config_i2p, i2p_port, i2p_location))
+
+# we can always create a Provider. If foolscap.connections.i2p or txi2p
+# are not installed, then get_i2p_handler() will return None. If tahoe.cfg
+# wants to start an I2P Destination too, then check_dest_config() will throw
+# a nice error, and startService will throw an ugly error.
+
+class Provider(service.MultiService):
+    def __init__(self, basedir, node_for_config, reactor):
+        service.MultiService.__init__(self)
+        self._basedir = basedir
+        self._node_for_config = node_for_config
+        self._i2p = _import_i2p()
+        self._txi2p = _import_txi2p()
+        self._reactor = reactor
+
+    def _get_i2p_config(self, *args, **kwargs):
+        return self._node_for_config.get_config("i2p", *args, **kwargs)
+
+    def get_i2p_handler(self):
+        enabled = self._get_i2p_config("enabled", True, boolean=True)
+        if not enabled:
+            return None
+        if not self._i2p:
+            return None
+
+        sam_port = self._get_i2p_config("sam.port", None)
+        launch = self._get_i2p_config("launch", False, boolean=True)
+        configdir = self._get_i2p_config("i2p.configdir", None)
+
+        if sam_port:
+            if launch:
+                raise ValueError("tahoe.cfg [i2p] must not set both "
+                                 "sam.port and launch")
+            ep = clientFromString(self._reactor, sam_port)
+            return self._i2p.sam_endpoint(ep)
+
+        if launch:
+            executable = self._get_i2p_config("i2p.executable", None)
+            return self._i2p.launch(i2p_configdir=configdir, i2p_binary=executable)
+
+        if configdir:
+            return self._i2p.local_i2p(configdir)
+
+        return self._i2p.default(self._reactor)
+
+    def check_dest_config(self):
+        if self._get_i2p_config("dest", False, boolean=True):
+            if not self._txi2p:
+                raise ValueError("Cannot create I2P Destination without txi2p. "
+                                 "Please 'pip install tahoe-lafs[i2p]' to fix.")
+
+            # to start an I2P server, we either need an I2P SAM port, or
+            # we need to launch I2P
+            sam_port = self._get_i2p_config("sam.port", None)
+            launch = self._get_i2p_config("launch", False, boolean=True)
+            configdir = self._get_i2p_config("i2p.configdir", None)
+            if not sam_port and not launch and not configdir:
+                raise ValueError("[i2p] dest = true, but we have neither "
+                                 "sam.port= nor launch=true nor configdir=")
+            if sam_port and launch:
+                raise ValueError("tahoe.cfg [i2p] must not set both "
+                                 "sam.port and launch")
+            if launch:
+                raise NotImplementedError("[i2p] launch is under development.")
+            # check that all the expected Destination-specific keys are present
+            def require(name):
+                if not self._get_i2p_config("dest.%s" % name, None):
+                    raise ValueError("[i2p] dest = true,"
+                                     " but dest.%s= is missing" % name)
+            require("port")
+            require("private_key_file")
+
+    def startService(self):
+        service.MultiService.startService(self)
+        # if we need to start I2P, now is the time
+        # TODO: implement i2p launching
+
+    @inlineCallbacks
+    def stopService(self):
+        # TODO: can we also stop i2p?
+        yield service.MultiService.stopService(self)