diff --git a/mypy.ini b/mypy.ini
index 482fd6dd8..901304c87 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -9,7 +9,7 @@ no_implicit_optional = True
 warn_redundant_casts = True
 strict_equality = True
 
-[mypy-allmydata.test.cli.wormholetesting,allmydata.test.test_connection_status]
+[mypy-allmydata.test.cli.wormholetesting,allmydata.listeners,allmydata.test.test_connection_status]
 disallow_any_generics = True
 disallow_subclassing_any = True
 disallow_untyped_calls = True
diff --git a/newsfragments/4004.minor b/newsfragments/4004.minor
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/allmydata/listeners.py b/src/allmydata/listeners.py
new file mode 100644
index 000000000..f97f699b4
--- /dev/null
+++ b/src/allmydata/listeners.py
@@ -0,0 +1,121 @@
+"""
+Define a protocol for listening on a transport such that Tahoe-LAFS can
+communicate over it, manage configuration for it in its configuration file,
+detect when it is possible to use it, etc.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Protocol, Sequence, Mapping, Optional, Union, Awaitable
+from typing_extensions import Literal
+
+from attrs import frozen
+from twisted.python.usage import Options
+
+from .interfaces import IAddressFamily
+from .util.iputil import allocate_tcp_port
+from .node import _Config
+
+@frozen
+class ListenerConfig:
+    """
+    :ivar tub_ports: Entries to merge into ``[node]tub.port``.
+
+    :ivar tub_locations: Entries to merge into ``[node]tub.location``.
+
+    :ivar node_config: Entries to add into the overall Tahoe-LAFS
+        configuration beneath a section named after this listener.
+    """
+    tub_ports: Sequence[str]
+    tub_locations: Sequence[str]
+    node_config: Mapping[str, Sequence[tuple[str, str]]]
+
+class Listener(Protocol):
+    """
+    An object which can listen on a transport and allow Tahoe-LAFS
+    communication to happen over it.
+    """
+    def is_available(self) -> bool:
+        """
+        Can this type of listener actually be used in this runtime
+        environment?
+        """
+
+    def can_hide_ip(self) -> bool:
+        """
+        Can the transport supported by this type of listener conceal the
+        node's public internet address from peers?
+        """
+
+    async def create_config(self, reactor: Any, cli_config: Options) -> Optional[ListenerConfig]:
+        """
+        Set up an instance of this listener according to the given
+        configuration parameters.
+
+        This may also allocate ephemeral resources if necessary.
+
+        :return: The created configuration which can be merged into the
+            overall *tahoe.cfg* configuration file.
+        """
+
+    def create(self, reactor: Any, config: _Config) -> IAddressFamily:
+        """
+        Instantiate this listener according to the given
+        previously-generated configuration.
+
+        :return: A handle on the listener which can be used to integrate it
+            into the Tahoe-LAFS node.
+        """
+
+class TCPProvider:
+    """
+    Support plain TCP connections.
+    """
+    def is_available(self) -> Literal[True]:
+        return True
+
+    def can_hide_ip(self) -> Literal[False]:
+        return False
+
+    async def create_config(self, reactor: Any, cli_config: Options) -> ListenerConfig:
+        tub_ports = []
+        tub_locations = []
+        if cli_config["port"]: # --port/--location are a pair
+            tub_ports.append(cli_config["port"])
+            tub_locations.append(cli_config["location"])
+        else:
+            assert "hostname" in cli_config
+            hostname = cli_config["hostname"]
+            new_port = allocate_tcp_port()
+            tub_ports.append(f"tcp:{new_port}")
+            tub_locations.append(f"tcp:{hostname}:{new_port}")
+
+        return ListenerConfig(tub_ports, tub_locations, {})
+
+    def create(self, reactor: Any, config: _Config) -> IAddressFamily:
+        raise NotImplementedError()
+
+
+@frozen
+class StaticProvider:
+    """
+    A provider that uses all pre-computed values.
+    """
+    _available: bool
+    _hide_ip: bool
+    _config: Union[Awaitable[Optional[ListenerConfig]], Optional[ListenerConfig]]
+    _address: IAddressFamily
+
+    def is_available(self) -> bool:
+        return self._available
+
+    def can_hide_ip(self) -> bool:
+        return self._hide_ip
+
+    async def create_config(self, reactor: Any, cli_config: Options) -> Optional[ListenerConfig]:
+        if self._config is None or isinstance(self._config, ListenerConfig):
+            return self._config
+        return await self._config
+
+    def create(self, reactor: Any, config: _Config) -> IAddressFamily:
+        return self._address
diff --git a/src/allmydata/scripts/create_node.py b/src/allmydata/scripts/create_node.py
index 7d15b95ec..4357abb49 100644
--- a/src/allmydata/scripts/create_node.py
+++ b/src/allmydata/scripts/create_node.py
@@ -1,3 +1,8 @@
+
+from __future__ import annotations
+
+from typing import Optional
+
 import io
 import os
 
@@ -19,9 +24,40 @@ from allmydata.scripts.common import (
     write_introducer,
 )
 from allmydata.scripts.default_nodedir import _default_nodedir
+from allmydata.util import dictutil
 from allmydata.util.assertutil import precondition
 from allmydata.util.encodingutil import listdir_unicode, argv_to_unicode, quote_local_unicode_path, get_io_encoding
-from allmydata.util import fileutil, i2p_provider, iputil, tor_provider, jsonbytes as json
+
+i2p_provider: Listener
+tor_provider: Listener
+
+from allmydata.util import fileutil, i2p_provider, tor_provider, jsonbytes as json
+
+from ..listeners import ListenerConfig, Listener, TCPProvider, StaticProvider
+
+def _get_listeners() -> dict[str, Listener]:
+    """
+    Get all of the kinds of listeners we might be able to use.
+    """
+    return {
+        "tor": tor_provider,
+        "i2p": i2p_provider,
+        "tcp": TCPProvider(),
+        "none": StaticProvider(
+            available=True,
+            hide_ip=False,
+            config=defer.succeed(None),
+            # This is supposed to be an IAddressFamily but we have none for
+            # this kind of provider.  We could implement new client and server
+            # endpoint types that always fail and pass an IAddressFamily here
+            # that uses those.  Nothing would ever even ask for them (at
+            # least, yet), let alone try to use them, so that's a lot of extra
+            # work for no practical result so I'm not doing it now.
+            address=None, # type: ignore[arg-type]
+        ),
+    }
+
+_LISTENERS = _get_listeners()
 
 dummy_tac = """
 import sys
@@ -98,8 +134,11 @@ def validate_where_options(o):
         if o['listen'] != "none" and o.get('join', None) is None:
             listeners = o['listen'].split(",")
             for l in listeners:
-                if l not in ["tcp", "tor", "i2p"]:
-                    raise UsageError("--listen= must be none, or one/some of: tcp, tor, i2p")
+                if l not in _LISTENERS:
+                    raise UsageError(
+                        "--listen= must be one/some of: "
+                        f"{', '.join(sorted(_LISTENERS))}",
+                    )
             if 'tcp' in listeners and not o['hostname']:
                 raise UsageError("--listen=tcp requires --hostname=")
             if 'tcp' not in listeners and o['hostname']:
@@ -108,7 +147,7 @@ def validate_where_options(o):
 def validate_tor_options(o):
     use_tor = "tor" in o["listen"].split(",")
     if use_tor or any((o["tor-launch"], o["tor-control-port"])):
-        if tor_provider._import_txtorcon() is None:
+        if not _LISTENERS["tor"].is_available():
             raise UsageError(
                 "Specifying any Tor options requires the 'txtorcon' module"
             )
@@ -123,7 +162,7 @@ def validate_tor_options(o):
 def validate_i2p_options(o):
     use_i2p = "i2p" in o["listen"].split(",")
     if use_i2p or any((o["i2p-launch"], o["i2p-sam-port"])):
-        if i2p_provider._import_txi2p() is None:
+        if not _LISTENERS["i2p"].is_available():
             raise UsageError(
                 "Specifying any I2P options requires the 'txi2p' module"
             )
@@ -145,11 +184,17 @@ class _CreateBaseOptions(BasedirOptions):
     def postOptions(self):
         super(_CreateBaseOptions, self).postOptions()
         if self['hide-ip']:
-            if tor_provider._import_txtorcon() is None and i2p_provider._import_txi2p() is None:
+            ip_hiders = dictutil.filter(lambda v: v.can_hide_ip(), _LISTENERS)
+            available = dictutil.filter(lambda v: v.is_available(), ip_hiders)
+            if not available:
                 raise UsageError(
-                    "--hide-ip was specified but neither 'txtorcon' nor 'txi2p' "
-                    "are installed.\nTo do so:\n   pip install tahoe-lafs[tor]\nor\n"
-                    "   pip install tahoe-lafs[i2p]"
+                    "--hide-ip was specified but no IP-hiding listener is installed.\n"
+                    "Try one of these:\n" +
+                    "".join([
+                        f"\tpip install tahoe-lafs[{name}]\n"
+                        for name
+                        in ip_hiders
+                    ])
                 )
 
 class CreateClientOptions(_CreateBaseOptions):
@@ -218,8 +263,34 @@ class CreateIntroducerOptions(NoDefaultBasedirOptions):
         validate_i2p_options(self)
 
 
-@defer.inlineCallbacks
-def write_node_config(c, config):
+def merge_config(
+        left: Optional[ListenerConfig],
+        right: Optional[ListenerConfig],
+) -> Optional[ListenerConfig]:
+    """
+    Merge two listener configurations into one configuration representing
+    both of them.
+
+    If either is ``None`` then the result is ``None``.  This supports the
+    "disable listeners" functionality.
+
+    :raise ValueError: If the keys in the node configs overlap.
+    """
+    if left is None or right is None:
+        return None
+
+    overlap = set(left.node_config) & set(right.node_config)
+    if overlap:
+        raise ValueError(f"Node configs overlap: {overlap}")
+
+    return ListenerConfig(
+        list(left.tub_ports) + list(right.tub_ports),
+        list(left.tub_locations) + list(right.tub_locations),
+        dict(list(left.node_config.items()) + list(right.node_config.items())),
+    )
+
+
+async def write_node_config(c, config):
     # this is shared between clients and introducers
     c.write("# -*- mode: conf; coding: {c.encoding} -*-\n".format(c=c))
     c.write("\n")
@@ -232,9 +303,10 @@ def write_node_config(c, config):
 
     if config["hide-ip"]:
         c.write("[connections]\n")
-        if tor_provider._import_txtorcon():
+        if _LISTENERS["tor"].is_available():
             c.write("tcp = tor\n")
         else:
+            # XXX What about i2p?
             c.write("tcp = disabled\n")
         c.write("\n")
 
@@ -253,38 +325,23 @@ def write_node_config(c, config):
     c.write("web.port = %s\n" % (webport,))
     c.write("web.static = public_html\n")
 
-    listeners = config['listen'].split(",")
+    listener_config = ListenerConfig([], [], {})
+    for listener_name in config['listen'].split(","):
+        listener = _LISTENERS[listener_name]
+        listener_config = merge_config(
+            (await listener.create_config(reactor, config)),
+            listener_config,
+        )
 
-    tor_config = {}
-    i2p_config = {}
-    tub_ports = []
-    tub_locations = []
-    if listeners == ["none"]:
-        c.write("tub.port = disabled\n")
-        c.write("tub.location = disabled\n")
+    if listener_config is None:
+        tub_ports = ["disabled"]
+        tub_locations = ["disabled"]
     else:
-        if "tor" in listeners:
-            (tor_config, tor_port, tor_location) = \
-                         yield tor_provider.create_config(reactor, config)
-            tub_ports.append(tor_port)
-            tub_locations.append(tor_location)
-        if "i2p" in listeners:
-            (i2p_config, i2p_port, i2p_location) = \
-                         yield i2p_provider.create_config(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"])
-                tub_locations.append(config["location"])
-            else:
-                assert "hostname" in config
-                hostname = config["hostname"]
-                new_port = iputil.allocate_tcp_port()
-                tub_ports.append("tcp:%s" % new_port)
-                tub_locations.append("tcp:%s:%s" % (hostname, new_port))
-        c.write("tub.port = %s\n" % ",".join(tub_ports))
-        c.write("tub.location = %s\n" % ",".join(tub_locations))
+        tub_ports = listener_config.tub_ports
+        tub_locations = listener_config.tub_locations
+
+    c.write("tub.port = %s\n" % ",".join(tub_ports))
+    c.write("tub.location = %s\n" % ",".join(tub_locations))
     c.write("\n")
 
     c.write("#log_gatherer.furl =\n")
@@ -294,17 +351,12 @@ def write_node_config(c, config):
     c.write("#ssh.authorized_keys_file = ~/.ssh/authorized_keys\n")
     c.write("\n")
 
-    if tor_config:
-        c.write("[tor]\n")
-        for key, value in list(tor_config.items()):
-            c.write("%s = %s\n" % (key, value))
-        c.write("\n")
-
-    if i2p_config:
-        c.write("[i2p]\n")
-        for key, value in list(i2p_config.items()):
-            c.write("%s = %s\n" % (key, value))
-        c.write("\n")
+    if listener_config is not None:
+        for section, items in listener_config.node_config.items():
+            c.write(f"[{section}]\n")
+            for k, v in items:
+                c.write(f"{k} = {v}\n")
+            c.write("\n")
 
 
 def write_client_config(c, config):
@@ -445,7 +497,7 @@ def create_node(config):
     fileutil.make_dirs(os.path.join(basedir, "private"), 0o700)
     cfg_name = os.path.join(basedir, "tahoe.cfg")
     with io.open(cfg_name, "w", encoding='utf-8') as c:
-        yield write_node_config(c, config)
+        yield defer.Deferred.fromCoroutine(write_node_config(c, config))
         write_client_config(c, config)
 
     print("Node created in %s" % quote_local_unicode_path(basedir), file=out)
@@ -488,7 +540,7 @@ def create_introducer(config):
     fileutil.make_dirs(os.path.join(basedir, "private"), 0o700)
     cfg_name = os.path.join(basedir, "tahoe.cfg")
     with io.open(cfg_name, "w", encoding='utf-8') as c:
-        yield write_node_config(c, config)
+        yield defer.Deferred.fromCoroutine(write_node_config(c, config))
 
     print("Introducer created in %s" % quote_local_unicode_path(basedir), file=out)
     defer.returnValue(0)
diff --git a/src/allmydata/test/cli/test_create.py b/src/allmydata/test/cli/test_create.py
index 1d1576082..406aebd48 100644
--- a/src/allmydata/test/cli/test_create.py
+++ b/src/allmydata/test/cli/test_create.py
@@ -17,6 +17,7 @@ from ..common import (
     disable_modules,
 )
 from ...scripts import create_node
+from ...listeners import ListenerConfig, StaticProvider
 from ... import client
 
 def read_config(basedir):
@@ -24,6 +25,68 @@ def read_config(basedir):
     config = configutil.get_config(tahoe_cfg)
     return config
 
+class MergeConfigTests(unittest.TestCase):
+    """
+    Tests for ``create_node.merge_config``.
+    """
+    def test_disable_left(self) -> None:
+        """
+        If the left argument to ``create_node.merge_config`` is ``None``
+        then the return value is ``None``.
+        """
+        conf = ListenerConfig([], [], {})
+        self.assertEqual(None, create_node.merge_config(None, conf))
+
+    def test_disable_right(self) -> None:
+        """
+        If the right argument to ``create_node.merge_config`` is ``None``
+        then the return value is ``None``.
+        """
+        conf = ListenerConfig([], [], {})
+        self.assertEqual(None, create_node.merge_config(conf, None))
+
+    def test_disable_both(self) -> None:
+        """
+        If both arguments to ``create_node.merge_config`` are ``None``
+        then the return value is ``None``.
+        """
+        self.assertEqual(None, create_node.merge_config(None, None))
+
+    def test_overlapping_keys(self) -> None:
+        """
+        If there are any keys in the ``node_config`` of the left and right
+        parameters that are shared then ``ValueError`` is raised.
+        """
+        left = ListenerConfig([], [], {"foo": [("b", "ar")]})
+        right = ListenerConfig([], [], {"foo": [("ba", "z")]})
+        self.assertRaises(ValueError, lambda: create_node.merge_config(left, right))
+
+    def test_merge(self) -> None:
+        """
+        ``create_node.merge_config`` returns a ``ListenerConfig`` that has
+        all of the ports, locations, and node config from each of the two
+        ``ListenerConfig`` values given.
+        """
+        left = ListenerConfig(
+            ["left-port"],
+            ["left-location"],
+            {"left": [("f", "oo")]},
+        )
+        right = ListenerConfig(
+            ["right-port"],
+            ["right-location"],
+            {"right": [("ba", "r")]},
+        )
+        result = create_node.merge_config(left, right)
+        self.assertEqual(
+            ListenerConfig(
+                ["left-port", "right-port"],
+                ["left-location", "right-location"],
+                {"left": [("f", "oo")], "right": [("ba", "r")]},
+            ),
+            result,
+        )
+
 class Config(unittest.TestCase):
     def test_client_unrecognized_options(self):
         tests = [
@@ -45,7 +108,14 @@ class Config(unittest.TestCase):
             e = self.assertRaises(usage.UsageError, parse_cli, verb, *args)
             self.assertIn("option %s not recognized" % (option,), str(e))
 
-    def test_create_client_config(self):
+    async def test_create_client_config(self):
+        """
+        ``create_node.write_client_config`` writes a configuration file
+        that can be parsed.
+
+        TODO Maybe we should test that we can recover the given configuration
+        from the parse, too.
+        """
         d = self.mktemp()
         os.mkdir(d)
         fname = os.path.join(d, 'tahoe.cfg')
@@ -59,7 +129,7 @@ class Config(unittest.TestCase):
                     "shares-happy": "1",
                     "shares-total": "1",
                     }
-            create_node.write_node_config(f, opts)
+            await create_node.write_node_config(f, opts)
             create_node.write_client_config(f, opts)
 
         # should succeed, no exceptions
@@ -245,7 +315,7 @@ class Config(unittest.TestCase):
                               parse_cli,
                               "create-node", "--listen=tcp,none",
                               basedir)
-        self.assertEqual(str(e), "--listen= must be none, or one/some of: tcp, tor, i2p")
+        self.assertEqual(str(e), "--listen=tcp requires --hostname=")
 
     def test_node_listen_bad(self):
         basedir = self.mktemp()
@@ -253,7 +323,7 @@ class Config(unittest.TestCase):
                               parse_cli,
                               "create-node", "--listen=XYZZY,tcp",
                               basedir)
-        self.assertEqual(str(e), "--listen= must be none, or one/some of: tcp, tor, i2p")
+        self.assertEqual(str(e), "--listen= must be one/some of: i2p, none, tcp, tor")
 
     def test_node_listen_tor_hostname(self):
         e = self.assertRaises(usage.UsageError,
@@ -287,24 +357,19 @@ class Config(unittest.TestCase):
         self.assertIn("To avoid clobbering anything, I am going to quit now", err)
 
     @defer.inlineCallbacks
-    def test_node_slow_tor(self):
-        basedir = self.mktemp()
+    def test_node_slow(self):
+        """
+        A node can be created using a listener type that returns an
+        unfired Deferred from its ``create_config`` method.
+        """
         d = defer.Deferred()
-        self.patch(tor_provider, "create_config", lambda *a, **kw: d)
-        d2 = run_cli("create-node", "--listen=tor", basedir)
-        d.callback(({}, "port", "location"))
-        rc, out, err = yield d2
-        self.assertEqual(rc, 0)
-        self.assertIn("Node created", out)
-        self.assertEqual(err, "")
+        slow = StaticProvider(True, False, d, None)
+        create_node._LISTENERS["xxyzy"] = slow
+        self.addCleanup(lambda: create_node._LISTENERS.pop("xxyzy"))
 
-    @defer.inlineCallbacks
-    def test_node_slow_i2p(self):
         basedir = self.mktemp()
-        d = defer.Deferred()
-        self.patch(i2p_provider, "create_config", lambda *a, **kw: d)
-        d2 = run_cli("create-node", "--listen=i2p", basedir)
-        d.callback(({}, "port", "location"))
+        d2 = run_cli("create-node", "--listen=xxyzy", basedir)
+        d.callback(None)
         rc, out, err = yield d2
         self.assertEqual(rc, 0)
         self.assertIn("Node created", out)
@@ -369,10 +434,12 @@ def fake_config(testcase: unittest.TestCase, module: Any, result: Any) -> list[t
 class Tor(unittest.TestCase):
     def test_default(self):
         basedir = self.mktemp()
-        tor_config = {"abc": "def"}
+        tor_config = {"tor": [("abc", "def")]}
         tor_port = "ghi"
         tor_location = "jkl"
-        config_d = defer.succeed( (tor_config, tor_port, tor_location) )
+        config_d = defer.succeed(
+            ListenerConfig([tor_port], [tor_location], tor_config)
+        )
 
         calls = fake_config(self, tor_provider, config_d)
         rc, out, err = self.successResultOf(
@@ -390,11 +457,12 @@ class Tor(unittest.TestCase):
         self.assertEqual(cfg.get("node", "tub.location"), "jkl")
 
     def test_launch(self):
+        """
+        The ``--tor-launch`` command line option sets ``tor-launch`` to
+        ``True``.
+        """
         basedir = self.mktemp()
-        tor_config = {"abc": "def"}
-        tor_port = "ghi"
-        tor_location = "jkl"
-        config_d = defer.succeed( (tor_config, tor_port, tor_location) )
+        config_d = defer.succeed(None)
 
         calls = fake_config(self, tor_provider, config_d)
         rc, out, err = self.successResultOf(
@@ -409,11 +477,12 @@ class Tor(unittest.TestCase):
         self.assertEqual(args[1]["tor-control-port"], None)
 
     def test_control_port(self):
+        """
+        The ``--tor-control-port`` command line parameter's value is
+        passed along as the ``tor-control-port`` value.
+        """
         basedir = self.mktemp()
-        tor_config = {"abc": "def"}
-        tor_port = "ghi"
-        tor_location = "jkl"
-        config_d = defer.succeed( (tor_config, tor_port, tor_location) )
+        config_d = defer.succeed(None)
 
         calls = fake_config(self, tor_provider, config_d)
         rc, out, err = self.successResultOf(
@@ -451,10 +520,10 @@ class Tor(unittest.TestCase):
 class I2P(unittest.TestCase):
     def test_default(self):
         basedir = self.mktemp()
-        i2p_config = {"abc": "def"}
+        i2p_config = {"i2p": [("abc", "def")]}
         i2p_port = "ghi"
         i2p_location = "jkl"
-        dest_d = defer.succeed( (i2p_config, i2p_port, i2p_location) )
+        dest_d = defer.succeed(ListenerConfig([i2p_port], [i2p_location], i2p_config))
 
         calls = fake_config(self, i2p_provider, dest_d)
         rc, out, err = self.successResultOf(
@@ -479,10 +548,7 @@ class I2P(unittest.TestCase):
 
     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) )
+        dest_d = defer.succeed(None)
 
         calls = fake_config(self, i2p_provider, dest_d)
         rc, out, err = self.successResultOf(
diff --git a/src/allmydata/test/test_dictutil.py b/src/allmydata/test/test_dictutil.py
index 7e26a6ed9..ce1c4a74c 100644
--- a/src/allmydata/test/test_dictutil.py
+++ b/src/allmydata/test/test_dictutil.py
@@ -1,17 +1,9 @@
 """
 Tests for allmydata.util.dictutil.
-
-Ported to Python 3.
 """
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-from __future__ import unicode_literals
+from __future__ import annotations
 
 from future.utils import PY2, PY3
-if PY2:
-    # dict omitted to match dictutil.py.
-    from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, list, object, range, str, max, min  # noqa: F401
 
 from unittest import skipIf
 
@@ -168,3 +160,18 @@ class TypedKeyDictPython2(unittest.TestCase):
         # Demonstration of how bytes and unicode can be mixed:
         d = {u"abc": 1}
         self.assertEqual(d[b"abc"], 1)
+
+
+class FilterTests(unittest.TestCase):
+    """
+    Tests for ``dictutil.filter``.
+    """
+    def test_filter(self) -> None:
+        """
+        ``dictutil.filter`` returns a ``dict`` that contains the key/value
+        pairs for which the value is matched by the given predicate.
+        """
+        self.assertEqual(
+            {1: 2},
+            dictutil.filter(lambda v: v == 2, {1: 2, 2: 3}),
+        )
diff --git a/src/allmydata/test/test_i2p_provider.py b/src/allmydata/test/test_i2p_provider.py
index 364a85c5b..072b17647 100644
--- a/src/allmydata/test/test_i2p_provider.py
+++ b/src/allmydata/test/test_i2p_provider.py
@@ -177,7 +177,7 @@ class CreateDest(unittest.TestCase):
                 with mock.patch("allmydata.util.i2p_provider.clientFromString",
                                 return_value=ep) as cfs:
                     d = i2p_provider.create_config(reactor, cli_config)
-        tahoe_config_i2p, i2p_port, i2p_location = self.successResultOf(d)
+                    i2p_config = self.successResultOf(d)
 
         connect_to_i2p.assert_called_with(reactor, cli_config, txi2p)
         cfs.assert_called_with(reactor, "goodport")
@@ -189,9 +189,9 @@ class CreateDest(unittest.TestCase):
                     "dest.private_key_file": os.path.join("private",
                                                           "i2p_dest.privkey"),
                     }
-        self.assertEqual(tahoe_config_i2p, expected)
-        self.assertEqual(i2p_port, "listen:i2p")
-        self.assertEqual(i2p_location, "i2p:FOOBAR.b32.i2p:3457")
+        self.assertEqual(dict(i2p_config.node_config["i2p"]), expected)
+        self.assertEqual(i2p_config.tub_ports, ["listen:i2p"])
+        self.assertEqual(i2p_config.tub_locations, ["i2p:FOOBAR.b32.i2p:3457"])
 
 _None = object()
 class FakeConfig(dict):
diff --git a/src/allmydata/test/test_tor_provider.py b/src/allmydata/test/test_tor_provider.py
index 20f947d55..fc29e41e4 100644
--- a/src/allmydata/test/test_tor_provider.py
+++ b/src/allmydata/test/test_tor_provider.py
@@ -197,7 +197,7 @@ class CreateOnion(unittest.TestCase):
                 with mock.patch("allmydata.util.tor_provider.allocate_tcp_port",
                                 return_value=999999):
                     d = tor_provider.create_config(reactor, cli_config)
-        tahoe_config_tor, tor_port, tor_location = self.successResultOf(d)
+                    tor_config = self.successResultOf(d)
 
         launch_tor.assert_called_with(reactor, executable,
                                       os.path.abspath(private_dir), txtorcon)
@@ -214,10 +214,10 @@ class CreateOnion(unittest.TestCase):
                     }
         if executable:
             expected["tor.executable"] = executable
-        self.assertEqual(tahoe_config_tor, expected)
-        self.assertEqual(tor_port, "tcp:999999:interface=127.0.0.1")
-        self.assertEqual(tor_location, "tor:ONION.onion:3457")
-        fn = os.path.join(basedir, tahoe_config_tor["onion.private_key_file"])
+        self.assertEqual(dict(tor_config.node_config["tor"]), expected)
+        self.assertEqual(tor_config.tub_ports, ["tcp:999999:interface=127.0.0.1"])
+        self.assertEqual(tor_config.tub_locations, ["tor:ONION.onion:3457"])
+        fn = os.path.join(basedir, dict(tor_config.node_config["tor"])["onion.private_key_file"])
         with open(fn, "rb") as f:
             privkey = f.read()
         self.assertEqual(privkey, b"privkey")
@@ -251,7 +251,7 @@ class CreateOnion(unittest.TestCase):
                 with mock.patch("allmydata.util.tor_provider.allocate_tcp_port",
                                 return_value=999999):
                     d = tor_provider.create_config(reactor, cli_config)
-        tahoe_config_tor, tor_port, tor_location = self.successResultOf(d)
+                    tor_config = self.successResultOf(d)
 
         connect_to_tor.assert_called_with(reactor, cli_config, txtorcon)
         txtorcon.EphemeralHiddenService.assert_called_with("3457 127.0.0.1:999999")
@@ -265,10 +265,10 @@ class CreateOnion(unittest.TestCase):
                     "onion.private_key_file": os.path.join("private",
                                                            "tor_onion.privkey"),
                     }
-        self.assertEqual(tahoe_config_tor, expected)
-        self.assertEqual(tor_port, "tcp:999999:interface=127.0.0.1")
-        self.assertEqual(tor_location, "tor:ONION.onion:3457")
-        fn = os.path.join(basedir, tahoe_config_tor["onion.private_key_file"])
+        self.assertEqual(dict(tor_config.node_config["tor"]), expected)
+        self.assertEqual(tor_config.tub_ports, ["tcp:999999:interface=127.0.0.1"])
+        self.assertEqual(tor_config.tub_locations, ["tor:ONION.onion:3457"])
+        fn = os.path.join(basedir, dict(tor_config.node_config["tor"])["onion.private_key_file"])
         with open(fn, "rb") as f:
             privkey = f.read()
         self.assertEqual(privkey, b"privkey")
diff --git a/src/allmydata/util/dictutil.py b/src/allmydata/util/dictutil.py
index 0a7df0a38..58820993f 100644
--- a/src/allmydata/util/dictutil.py
+++ b/src/allmydata/util/dictutil.py
@@ -2,6 +2,23 @@
 Tools to mess with dicts.
 """
 
+from __future__ import annotations
+from typing import Callable, TypeVar
+
+K = TypeVar("K")
+V = TypeVar("V")
+
+def filter(pred: Callable[[V], bool], orig: dict[K, V]) -> dict[K, V]:
+    """
+    Filter out key/value pairs whose value fails to match a predicate.
+    """
+    return {
+        k: v
+        for (k, v)
+        in orig.items()
+        if pred(v)
+    }
+
 class DictOfSets(dict):
     def add(self, key, value):
         if key in self:
diff --git a/src/allmydata/util/i2p_provider.py b/src/allmydata/util/i2p_provider.py
index 071245adf..c480cd2f1 100644
--- a/src/allmydata/util/i2p_provider.py
+++ b/src/allmydata/util/i2p_provider.py
@@ -1,14 +1,9 @@
 # -*- coding: utf-8 -*-
-"""
-Ported to Python 3.
-"""
-from __future__ import absolute_import, print_function, with_statement
-from __future__ import division
-from __future__ import unicode_literals
 
-from future.utils import PY2
-if PY2:
-    from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min  # noqa: F401
+from __future__ import annotations
+
+from typing import Any
+from typing_extensions import Literal
 
 import os
 
@@ -20,12 +15,15 @@ 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
+from twisted.python.usage import Options
 
+from ..listeners import ListenerConfig
 from ..interfaces import (
     IAddressFamily,
 )
+from ..node import _Config
 
-def create(reactor, config):
+def create(reactor: Any, config: _Config) -> IAddressFamily:
     """
     Create a new Provider service (this is an IService so must be
     hooked up to a parent or otherwise started).
@@ -55,6 +53,21 @@ def _import_txi2p():
     except ImportError: # pragma: no cover
         return None
 
+def is_available() -> bool:
+    """
+    Can this type of listener actually be used in this runtime
+    environment?
+
+    If its dependencies are missing then it cannot be.
+    """
+    return not (_import_i2p() is None or _import_txi2p() is None)
+
+def can_hide_ip() -> Literal[True]:
+    """
+    Can the transport supported by this type of listener conceal the
+    node's public internet address from peers?
+    """
+    return True
 
 def _try_to_connect(reactor, endpoint_desc, stdout, txi2p):
     # yields True or None
@@ -97,29 +110,35 @@ def _connect_to_i2p(reactor, cli_config, txi2p):
     else:
         raise ValueError("unable to reach any default I2P SAM port")
 
-@inlineCallbacks
-def create_config(reactor, cli_config):
+async def create_config(reactor: Any, cli_config: Options) -> ListenerConfig:
+    """
+    For a given set of command-line options, construct an I2P listener.
+
+    This includes allocating a new I2P address.
+    """
     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]
+    tahoe_config_i2p = [] # written into tahoe.cfg:[i2p]
     private_dir = os.path.abspath(os.path.join(cli_config["basedir"], "private"))
-    stdout = cli_config.stdout
+    # XXX We shouldn't carry stdout around by jamming it into the Options
+    # value.  See https://tahoe-lafs.org/trac/tahoe-lafs/ticket/4048
+    stdout = cli_config.stdout # type: ignore[attr-defined]
     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)
+        sam_port = await _connect_to_i2p(reactor, cli_config, txi2p)
         print("I2P connection established", file=stdout)
-        tahoe_config_i2p["sam.port"] = sam_port
+        tahoe_config_i2p.append(("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)
+    dest = await txi2p.generateDestination(reactor, privkeyfile, 'SAM', sam_endpoint)
     print(".i2p address allocated", file=stdout)
     i2p_port = "listen:i2p" # means "see [i2p]", calls Provider.get_listener()
     i2p_location = "i2p:%s:%d" % (dest.host, external_port)
@@ -132,10 +151,11 @@ def create_config(reactor, cli_config):
     # * "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.extend([
+        ("dest", "true"),
+        ("dest.port", str(external_port)),
+        ("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
@@ -149,7 +169,7 @@ def create_config(reactor, cli_config):
     # at both create-node and startup time. The data directory is not
     # recorded in tahoe.cfg
 
-    returnValue((tahoe_config_i2p, i2p_port, i2p_location))
+    return ListenerConfig([i2p_port], [i2p_location], {"i2p": tahoe_config_i2p})
 
 
 @implementer(IAddressFamily)
diff --git a/src/allmydata/util/iputil.py b/src/allmydata/util/iputil.py
index fd3e88c7f..e71e514e8 100644
--- a/src/allmydata/util/iputil.py
+++ b/src/allmydata/util/iputil.py
@@ -1,17 +1,10 @@
 """
 Utilities for getting IP addresses.
-
-Ported to Python 3.
 """
 
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-from __future__ import unicode_literals
+from future.utils import native_str
 
-from future.utils import PY2, native_str
-if PY2:
-    from builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min  # noqa: F401
+from typing import Callable
 
 import os, socket
 
@@ -39,6 +32,7 @@ from .gcutil import (
 
 fcntl = requireModule("fcntl")
 
+allocate_tcp_port: Callable[[], int]
 from foolscap.util import allocate_tcp_port # re-exported
 
 try:
diff --git a/src/allmydata/util/tor_provider.py b/src/allmydata/util/tor_provider.py
index aaf43db73..c40e65f42 100644
--- a/src/allmydata/util/tor_provider.py
+++ b/src/allmydata/util/tor_provider.py
@@ -1,11 +1,8 @@
 # -*- coding: utf-8 -*-
-"""
-Ported to Python 3.
-"""
-
 from __future__ import annotations
 
-from typing import Optional
+from typing import Any
+from typing_extensions import Literal
 import os
 
 from zope.interface import (
@@ -16,12 +13,14 @@ from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.internet.endpoints import clientFromString, TCP4ServerEndpoint
 from twisted.internet.error import ConnectionRefusedError, ConnectError
 from twisted.application import service
+from twisted.python.usage import Options
 
 from .observer import OneShotObserverList
 from .iputil import allocate_tcp_port
 from ..interfaces import (
     IAddressFamily,
 )
+from ..listeners import ListenerConfig
 
 
 def _import_tor():
@@ -38,7 +37,13 @@ def _import_txtorcon():
     except ImportError: # pragma: no cover
         return None
 
-def create(reactor, config, import_tor=None, import_txtorcon=None) -> Optional[_Provider]:
+def can_hide_ip() -> Literal[True]:
+    return True
+
+def is_available() -> bool:
+    return not (_import_tor() is None or _import_txtorcon() is None)
+
+def create(reactor, config, import_tor=None, import_txtorcon=None) -> _Provider:
     """
     Create a new _Provider service (this is an IService so must be
     hooked up to a parent or otherwise started).
@@ -150,31 +155,32 @@ def _connect_to_tor(reactor, cli_config, txtorcon):
     else:
         raise ValueError("unable to reach any default Tor control port")
 
-@inlineCallbacks
-def create_config(reactor, cli_config):
+async def create_config(reactor: Any, cli_config: Options) -> ListenerConfig:
     txtorcon = _import_txtorcon()
     if not txtorcon:
         raise ValueError("Cannot create onion without txtorcon. "
                          "Please 'pip install tahoe-lafs[tor]' to fix this.")
-    tahoe_config_tor = {} # written into tahoe.cfg:[tor]
+    tahoe_config_tor = [] # written into tahoe.cfg:[tor]
     private_dir = os.path.abspath(os.path.join(cli_config["basedir"], "private"))
-    stdout = cli_config.stdout
+    # XXX We shouldn't carry stdout around by jamming it into the Options
+    # value.  See https://tahoe-lafs.org/trac/tahoe-lafs/ticket/4048
+    stdout = cli_config.stdout # type: ignore[attr-defined]
     if cli_config["tor-launch"]:
-        tahoe_config_tor["launch"] = "true"
+        tahoe_config_tor.append(("launch", "true"))
         tor_executable = cli_config["tor-executable"]
         if tor_executable:
-            tahoe_config_tor["tor.executable"] = tor_executable
+            tahoe_config_tor.append(("tor.executable", tor_executable))
         print("launching Tor (to allocate .onion address)..", file=stdout)
-        (_, tor) = yield _launch_tor(
+        (_, tor) = await _launch_tor(
             reactor, tor_executable, private_dir, txtorcon)
         tor_control_proto = tor.protocol
         print("Tor launched", file=stdout)
     else:
         print("connecting to Tor (to allocate .onion address)..", file=stdout)
-        (port, tor_control_proto) = yield _connect_to_tor(
+        (port, tor_control_proto) = await _connect_to_tor(
             reactor, cli_config, txtorcon)
         print("Tor connection established", file=stdout)
-        tahoe_config_tor["control.port"] = port
+        tahoe_config_tor.append(("control.port", port))
 
     external_port = 3457 # TODO: pick this randomly? there's no contention.
 
@@ -183,12 +189,12 @@ def create_config(reactor, cli_config):
         "%d 127.0.0.1:%d" % (external_port, local_port)
     )
     print("allocating .onion address (takes ~40s)..", file=stdout)
-    yield ehs.add_to_tor(tor_control_proto)
+    await ehs.add_to_tor(tor_control_proto)
     print(".onion address allocated", file=stdout)
     tor_port = "tcp:%d:interface=127.0.0.1" % local_port
     tor_location = "tor:%s:%d" % (ehs.hostname, external_port)
     privkey = ehs.private_key
-    yield ehs.remove_from_tor(tor_control_proto)
+    await ehs.remove_from_tor(tor_control_proto)
 
     # in addition to the "how to launch/connect-to tor" keys above, we also
     # record information about the onion service into tahoe.cfg.
@@ -200,12 +206,12 @@ def create_config(reactor, cli_config):
     # * "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_tor["onion"] = "true"
-    tahoe_config_tor["onion.local_port"] = str(local_port)
-    tahoe_config_tor["onion.external_port"] = str(external_port)
-    assert privkey
-    tahoe_config_tor["onion.private_key_file"] = os.path.join("private",
-                                                              "tor_onion.privkey")
+    tahoe_config_tor.extend([
+        ("onion", "true"),
+        ("onion.local_port", str(local_port)),
+        ("onion.external_port", str(external_port)),
+        ("onion.private_key_file", os.path.join("private", "tor_onion.privkey")),
+    ])
     privkeyfile = os.path.join(private_dir, "tor_onion.privkey")
     with open(privkeyfile, "wb") as f:
         if isinstance(privkey, str):
@@ -224,7 +230,11 @@ def create_config(reactor, cli_config):
     # at both create-node and startup time. The data directory is not
     # recorded in tahoe.cfg
 
-    returnValue((tahoe_config_tor, tor_port, tor_location))
+    return ListenerConfig(
+        [tor_port],
+        [tor_location],
+        {"tor": tahoe_config_tor},
+    )
 
 
 @implementer(IAddressFamily)