add tor transport

4 files changed, 604 insertions, 5 deletions

diff --git a/src/config/mod.rs b/src/config/mod.rs
index 2de44ac..d4a7c9c 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -240,7 +240,7 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         Kind::List,
         "NAME,[NAME...]",
         Some("mycelium"),
-        "The transports this node uses, in order of preference: mycelium, libp2p (in builds with the libp2p feature). A peer is reached on the first one both have; nodes on several carry the traffic between nodes that share none.",
+        "The transports this node uses, in order of preference: mycelium, libp2p and tor (in builds with the libp2p and tor features). A peer is reached on the first one both have, until the paths are measured (then on the fastest); nodes on several carry the traffic between nodes that share none.",
     ),
     option(
         "libp2p-listen",
@@ -285,6 +285,20 @@ pub(crate) const OPTIONS: &[OptionSpec] = &[
         "With the libp2p transport: use public libp2p infrastructure: join the IPFS DHT (through the IPFS bootstrap nodes) to find peers' addresses and public relays, picked after members that relay. Public relays only keep circuits for minutes and some KiB: enough to punch holes through NATs. Makes this node's peer ID and addresses known to the DHT.",
     ),
     option(
+        "tor-state-dir",
+        Kind::Text,
+        "PATH",
+        None,
+        "With the tor transport: arti's state and cache (in `state` and `cache` below), among them the onion service's key, which makes its address: keep it with the node. Required with tor.",
+    ),
+    option(
+        "tor-port",
+        Kind::Integer,
+        "PORT",
+        Some("9662"),
+        "With the tor transport: the port of this node's onion service (published with its onion address).",
+    ),
+    option(
         "libp2p-key-file",
         Kind::Text,
         "PATH",
@@ -420,6 +434,11 @@ pub(crate) struct Settings {
     pub(crate) libp2p_public: bool,
     #[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
     pub(crate) libp2p_key_file: Option<PathBuf>,
+    #[cfg_attr(not(feature = "tor"), allow(dead_code))]
+    pub(crate) tor_state_dir: Option<PathBuf>,
+    #[serde(default = "default_tor_port")]
+    #[cfg_attr(not(feature = "tor"), allow(dead_code))]
+    pub(crate) tor_port: u16,
     pub(crate) mycelium_rest_url: Option<String>,
     pub(crate) mycelium_key_file: Option<PathBuf>,
     pub(crate) mycelium_network_key_file: Option<PathBuf>,
@@ -455,8 +474,14 @@ fn default_transports() -> Vec<String> {
     vec!["mycelium".to_owned()]
 }
 /// The transports `transports` may name, and whether this build has each.
-pub(crate) const TRANSPORTS: &[(&str, bool)] =
-    &[("mycelium", true), ("libp2p", cfg!(feature = "libp2p"))];
+pub(crate) const TRANSPORTS: &[(&str, bool)] = &[
+    ("mycelium", true),
+    ("libp2p", cfg!(feature = "libp2p")),
+    ("tor", cfg!(feature = "tor")),
+];
+fn default_tor_port() -> u16 {
+    9662
+}
 fn default_mycelium_peers() -> Vec<String> {
     vec!["tls://127.0.0.1:9651".to_owned()]
 }
@@ -521,6 +546,11 @@ impl Settings {
         if self.transports.is_empty() {
             bail!("transports: a node needs at least one");
         }
+        if self.transports.iter().any(|name| name == "tor") && self.tor_state_dir.is_none() {
+            bail!(
+                "tor needs tor-state-dir: where arti keeps its state, the onion service's key among it"
+            );
+        }
         for (index, name) in self.transports.iter().enumerate() {
             match TRANSPORTS.iter().find(|(known, _)| known == name) {
                 None => bail!(
diff --git a/src/main.rs b/src/main.rs
index f2494d8..a95d764 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -53,7 +53,10 @@ async fn main() -> Result<()> {
     let args = &config.settings;
 
     // Read before anything starts, so a bad key file fails right away.
-    let node_key = identity::NodeKey::load(&args.node_key_file, args.create_node_key)?;
+    let node_key = Arc::new(identity::NodeKey::load(
+        &args.node_key_file,
+        args.create_node_key,
+    )?);
     info!(node_id = %node_key.id(), "Node identity");
     let swarm_key = match &args.swarm_key_file {
         Some(path) => Some(Arc::new(swarm::auth::SwarmKey::from_file(path)?)),
@@ -222,6 +225,22 @@ async fn main() -> Result<()> {
                     )?),
                 )
             }
+            #[cfg(feature = "tor")]
+            "tor" => {
+                let state_dir = args
+                    .tor_state_dir
+                    .clone()
+                    .context("tor needs tor-state-dir")?;
+                let transport = tokio::select! {
+                    transport = transport::tor::TorTransport::start(transport::tor::TorConfig {
+                        key: Arc::clone(&node_key),
+                        state_dir,
+                        port: args.tor_port,
+                    }) => transport?,
+                    _ = stopped(stop.clone()) => return Ok(()),
+                };
+                ("tor", Arc::new(transport) as Arc<dyn Transport>)
+            }
             other => anyhow::bail!("transports: no {other} in this build"),
         };
         debug!(
@@ -250,7 +269,7 @@ async fn main() -> Result<()> {
     let node = node::Node::new(node::NodeConfig {
         transport,
         identity: Some(transport::identity::IdentityConfig {
-            key: Arc::new(node_key),
+            key: Arc::clone(&node_key),
             scheme,
             more: transports.collect(),
         }),
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index f28ac1f..15b286a 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -23,6 +23,8 @@ pub(crate) mod paths;
 #[cfg(test)]
 pub(crate) mod sim;
 pub(crate) mod stats;
+#[cfg(feature = "tor")]
+pub(crate) mod tor;
 pub(crate) mod traffic;
 pub(crate) mod wire;
 
diff --git a/src/transport/tor.rs b/src/transport/tor.rs
new file mode 100644
index 0000000..d4a5604
--- /dev/null
+++ b/src/transport/tor.rs
@@ -0,0 +1,548 @@
+//! Tor as a transport (`--transports tor`, in builds with the `tor` feature), for
+//! reachability: every node runs an onion service (arti), reachable from anywhere
+//! without a forwarded port or a relay. Not for anonymity: a node's other addresses are
+//! in the member list next to its onion address.
+//!
+//! A node's transport identity is its node key (64 hex digits); where it is, its onion
+//! service: `<onion>.onion:<port>/<key>` (`locate`, `locations`). The onion service's
+//! own key lives in arti's state (`--tor-state-dir`), so its address stays the same
+//! across restarts.
+//!
+//! Peers talk over one long-lived connection per pair (opening a stream over Tor costs
+//! a round trip through the rendezvous circuit): a Tor stream to the other's onion
+//! service, either way. Tor authenticates the onion service to the one who dials it,
+//! but nothing about the dialer to the service: the challenge handshake
+//! (`transport::challenge`) at the start of every connection proves both keys, bound to
+//! the onion address dialed. Then messages go as on libp2p: their length (4 bytes, big
+//! endian), then the message as `wire` encodes it. A connection that breaks is dialed
+//! again on the next send. Both ends may dial at once; each then sends on the newest
+//! connection and reads from both until the older one closes.
+//!
+//! Onion circuits are slow (round trips of a second or more, seconds to set one up),
+//! so the timeouts are generous and messages smaller than on libp2p; the paths layer
+//! (`transport::paths`) prefers faster transports while they work.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, MutexGuard};
+use std::time::Duration;
+
+use anyhow::{Context, Result, anyhow, bail};
+use arti_client::config::TorClientConfigBuilder;
+use arti_client::{DataStream, TorClient};
+use async_trait::async_trait;
+use futures::StreamExt;
+use safelog::DisplayRedacted as _;
+use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
+use tokio::sync::{mpsc, watch};
+use tor_cell::relaycell::msg::{Connected, End, EndReason};
+use tor_hsservice::config::OnionServiceConfigBuilder;
+use tor_hsservice::{RunningOnionService, StreamRequest, handle_rend_requests};
+use tor_proto::stream::IncomingStreamRequest;
+use tor_rtcompat::PreferredRuntime;
+use tracing::{debug, info, warn};
+
+use crate::identity::NodeKey;
+use crate::locks::MutexExt;
+use crate::transport::challenge::{self, Role};
+use crate::transport::{EncodedMessage, Hex, MessageHandler, ParsedMessage, Transport, wire};
+
+/// The onion service's nickname in arti's state (one per node).
+const NICKNAME: &str = "continuum";
+/// The largest message read. Sync messages above 16 chunks travel as blobs, and a chunk
+/// is at most 4 MiB.
+const MAX_MESSAGE: usize = 8 << 20;
+/// How long reaching an onion service may take (building circuits, the rendezvous).
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(60);
+/// How long the challenge handshake may take, a few round trips.
+const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
+/// How long a message may take, before what its size adds (`stream_timeout`).
+const STREAM_TIMEOUT: Duration = Duration::from_secs(30);
+/// The slowest a message is expected to go over Tor, in bytes per second.
+const MIN_RATE: u64 = 64 << 10;
+/// Messages read and not handled yet, at most.
+const READS: usize = 32;
+/// Peers with a known location, at most.
+const MAX_PEERS: usize = 4096;
+
+/// How long a message of `bytes` may take.
+fn stream_timeout(bytes: usize) -> Duration {
+    STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE)
+}
+
+/// What a Tor transport is made with.
+pub(crate) struct TorConfig {
+    /// The node key: it proves who we are in the challenge handshake.
+    pub(crate) key: Arc<NodeKey>,
+    /// Arti's state (the onion service's key, among others) and cache, in `state` and
+    /// `cache` below.
+    pub(crate) state_dir: PathBuf,
+    /// The onion service's port.
+    pub(crate) port: u16,
+}
+
+/// Whether `id` is a Tor transport identity: a public key, 64 lowercase hex digits.
+fn is_key(id: &str) -> bool {
+    id.len() == 64
+        && id
+            .bytes()
+            .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
+}
+
+/// A location, `<onion>.onion:<port>`, and the key there, if `target` names one.
+fn parse_location(target: &str) -> Option<(String, Option<String>)> {
+    let (location, key) = match target.split_once('/') {
+        Some((location, key)) => (location, Some(key)),
+        None => (target, None),
+    };
+    let (host, port) = location.rsplit_once(':')?;
+    let name = host.strip_suffix(".onion")?;
+    if name.is_empty() || !name.bytes().all(|byte| byte.is_ascii_alphanumeric()) {
+        return None;
+    }
+    port.parse::<u16>().ok()?;
+    if key.is_some_and(|key| !is_key(key)) {
+        return None;
+    }
+    Some((location.to_ascii_lowercase(), key.map(str::to_owned)))
+}
+
+/// One connection to a peer: where to write. A reader task reads the other half.
+struct Connection {
+    id: u64,
+    writer: tokio::sync::Mutex<WriteHalf<DataStream>>,
+}
+
+struct Shared {
+    key: Arc<NodeKey>,
+    local_id: String,
+    /// Where each peer's onion service is, by key.
+    book: Mutex<HashMap<String, String>>,
+    /// The connection to write to each peer on, by key.
+    connections: Mutex<HashMap<String, Arc<Connection>>>,
+    /// One dial at a time per location.
+    dialing: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+    next_connection: AtomicU64,
+    /// Messages read, with the sender's key, for the consumer.
+    inbound: mpsc::Sender<(String, Vec<u8>)>,
+}
+
+fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+    mutex.lock_or_recover()
+}
+
+pub(crate) struct TorTransport {
+    shared: Arc<Shared>,
+    client: Arc<TorClient<PreferredRuntime>>,
+    /// Our onion service, `<onion>.onion:<port>`.
+    onion: String,
+    /// Kept: dropping it stops the onion service.
+    _service: Arc<RunningOnionService>,
+    incoming: tokio::sync::Mutex<Option<mpsc::Receiver<(String, Vec<u8>)>>>,
+    stopped: watch::Receiver<bool>,
+}
+
+impl TorTransport {
+    /// Bootstraps arti (this takes a while: it fetches what it needs to know of the Tor
+    /// network) and launches our onion service.
+    pub(crate) async fn start(config: TorConfig) -> Result<Self> {
+        let tor_config = TorClientConfigBuilder::from_directories(
+            config.state_dir.join("state"),
+            config.state_dir.join("cache"),
+        )
+        .build()
+        .context("Configuring Tor")?;
+        info!(state_dir = %config.state_dir.display(), "Bootstrapping Tor");
+        let client = TorClient::create_bootstrapped(tor_config)
+            .await
+            .context("Bootstrapping Tor")?;
+        let service_config = OnionServiceConfigBuilder::default()
+            .nickname(
+                NICKNAME
+                    .parse()
+                    .map_err(|error| anyhow!("Onion service nickname: {error}"))?,
+            )
+            .build()
+            .context("Configuring the onion service")?;
+        let (service, requests) = client
+            .launch_onion_service(service_config)
+            .context("Launching the onion service")?
+            .ok_or_else(|| anyhow!("The onion service is disabled in arti's configuration"))?;
+        let name = service
+            .onion_address()
+            .ok_or_else(|| anyhow!("The onion service has no address"))?;
+        // Arti shows onion addresses redacted unless asked: we publish ours.
+        let onion = format!("{}:{}", name.display_unredacted(), config.port);
+
+        let local_id = Hex(&config.key.public()).to_string();
+        let (inbound, incoming) = mpsc::channel(READS);
+        let shared = Arc::new(Shared {
+            key: config.key,
+            local_id,
+            book: Mutex::default(),
+            connections: Mutex::default(),
+            dialing: Mutex::default(),
+            next_connection: AtomicU64::new(0),
+            inbound,
+        });
+        let (stopping, stopped) = watch::channel(false);
+        {
+            let shared = Arc::clone(&shared);
+            let onion = onion.clone();
+            let port = config.port;
+            tokio::spawn(async move {
+                let requests = handle_rend_requests(requests);
+                futures::pin_mut!(requests);
+                while let Some(request) = requests.next().await {
+                    let shared = Arc::clone(&shared);
+                    let onion = onion.clone();
+                    tokio::spawn(async move { shared.accept(request, &onion, port).await });
+                }
+                warn!("The onion service stopped");
+                let _ = stopping.send(true);
+            });
+        }
+        info!(address = %onion, "Tor transport started (the onion service may take a minute to be reachable)");
+        Ok(Self {
+            shared,
+            client,
+            onion,
+            _service: service,
+            incoming: tokio::sync::Mutex::new(Some(incoming)),
+            stopped,
+        })
+    }
+
+    /// A connection to the peer with `key`, made if there is none.
+    async fn connection(&self, key: &str) -> Result<Arc<Connection>> {
+        if let Some(connection) = lock(&self.shared.connections).get(key) {
+            return Ok(Arc::clone(connection));
+        }
+        let Some(location) = lock(&self.shared.book).get(key).cloned() else {
+            bail!("no onion address known for {key}");
+        };
+        self.dial(&location, Some(key)).await?;
+        lock(&self.shared.connections)
+            .get(key)
+            .cloned()
+            .ok_or_else(|| anyhow!("the connection to {key} closed at once"))
+    }
+
+    /// Dials the onion service at `location` and runs the handshake: returns the key
+    /// there, which must be `expected` if given.
+    async fn dial(&self, location: &str, expected: Option<&str>) -> Result<String> {
+        let dialing = Arc::clone(
+            lock(&self.shared.dialing)
+                .entry(location.to_owned())
+                .or_default(),
+        );
+        let _one_at_a_time = dialing.lock().await;
+        // Someone else may have connected meanwhile.
+        if let Some(expected) = expected
+            && lock(&self.shared.connections).contains_key(expected)
+        {
+            return Ok(expected.to_owned());
+        }
+        let result = async {
+            let (host, port) = location
+                .rsplit_once(':')
+                .and_then(|(host, port)| Some((host, port.parse::<u16>().ok()?)))
+                .ok_or_else(|| anyhow!("{location} is no onion address with a port"))?;
+            let mut stream =
+                tokio::time::timeout(CONNECT_TIMEOUT, self.client.connect((host, port)))
+                    .await
+                    .map_err(|_| {
+                        anyhow!("reaching {location} took longer than {CONNECT_TIMEOUT:?}")
+                    })?
+                    .with_context(|| format!("Reaching {location}"))?;
+            let key = challenge::run(
+                &mut stream,
+                &self.shared.key,
+                Role::Dialer,
+                location,
+                HANDSHAKE_TIMEOUT,
+            )
+            .await
+            .with_context(|| format!("Handshake with {location}"))?;
+            let key = Hex(&key).to_string();
+            if let Some(expected) = expected
+                && key != expected
+            {
+                bail!("{location} is another node now ({key}, not {expected})");
+            }
+            self.shared.remember(&key, location);
+            self.shared.register(key.clone(), stream);
+            Ok(key)
+        }
+        .await;
+        lock(&self.shared.dialing).remove(location);
+        result
+    }
+}
+
+impl Shared {
+    /// `key` is at `location`.
+    fn remember(&self, key: &str, location: &str) {
+        if key == self.local_id {
+            return;
+        }
+        let mut book = lock(&self.book);
+        if book.len() < MAX_PEERS || book.contains_key(key) {
+            book.insert(key.to_owned(), location.to_owned());
+        }
+    }
+
+    /// A stream to our onion service: accepted if for our port, then the handshake.
+    async fn accept(self: Arc<Self>, request: StreamRequest, onion: &str, port: u16) {
+        let ours = matches!(request.request(), IncomingStreamRequest::Begin(begin) if begin.port() == port);
+        if !ours {
+            let _ = request
+                .reject(End::new_with_reason(EndReason::CONNECTREFUSED))
+                .await;
+            return;
+        }
+        let mut stream = match request.accept(Connected::new_empty()).await {
+            Ok(stream) => stream,
+            Err(error) => {
+                debug!(error = %error, "Couldn't accept a stream to our onion service");
+                return;
+            }
+        };
+        match challenge::run(
+            &mut stream,
+            &self.key,
+            Role::Listener,
+            onion,
+            HANDSHAKE_TIMEOUT,
+        )
+        .await
+        {
+            Ok(key) => self.register(Hex(&key).to_string(), stream),
+            Err(error) => {
+                debug!(error = %format!("{error:#}"), "Dropping a connection to our onion service: the handshake failed");
+            }
+        }
+    }
+
+    /// A connection to `key`, authenticated: the one to write to it on from now, and
+    /// read from until it closes.
+    fn register(self: &Arc<Self>, key: String, stream: DataStream) {
+        let (reader, writer) = tokio::io::split(stream);
+        let id = self.next_connection.fetch_add(1, Ordering::Relaxed);
+        let connection = Arc::new(Connection {
+            id,
+            writer: tokio::sync::Mutex::new(writer),
+        });
+        lock(&self.connections).insert(key.clone(), connection);
+        debug!(peer = %key, connection = id, "Tor connection up");
+        let shared = Arc::clone(self);
+        tokio::spawn(async move {
+            let error = shared.read(&key, reader).await;
+            debug!(peer = %key, connection = id, error = %format!("{error:#}"), "Tor connection closed");
+            shared.forget(&key, id);
+        });
+    }
+
+    /// Reads messages from `key` until the connection fails; returns why.
+    async fn read(&self, key: &str, mut reader: ReadHalf<DataStream>) -> anyhow::Error {
+        loop {
+            let mut length = [0u8; 4];
+            if let Err(error) = reader.read_exact(&mut length).await {
+                return error.into();
+            }
+            let length = u32::from_be_bytes(length) as usize;
+            if length > MAX_MESSAGE {
+                return anyhow!("a message of {length} bytes; at most {MAX_MESSAGE} are read");
+            }
+            let mut body = vec![0u8; length];
+            match tokio::time::timeout(stream_timeout(length), reader.read_exact(&mut body)).await {
+                Ok(Ok(_)) => {}
+                Ok(Err(error)) => return error.into(),
+                Err(_) => {
+                    return anyhow!(
+                        "{length} bytes took longer than {:?}",
+                        stream_timeout(length)
+                    );
+                }
+            }
+            if self.inbound.send((key.to_owned(), body)).await.is_err() {
+                return anyhow!("the transport stopped");
+            }
+        }
+    }
+
+    /// The connection `id` to `key` is gone: not to be written to any more.
+    fn forget(&self, key: &str, id: u64) {
+        let mut connections = lock(&self.connections);
+        if connections
+            .get(key)
+            .is_some_and(|connection| connection.id == id)
+        {
+            connections.remove(key);
+        }
+    }
+}
+
+#[async_trait]
+impl Transport for TorTransport {
+    fn local_id(&self) -> &str {
+        &self.shared.local_id
+    }
+
+    fn is_peer_id(&self, id: &str) -> bool {
+        is_key(id)
+    }
+
+    /// The longest a message may take.
+    fn delivery_limit(&self) -> Duration {
+        stream_timeout(MAX_MESSAGE)
+    }
+
+    /// Our onion service, with our key.
+    fn local_addresses(&self) -> Vec<String> {
+        vec![format!("{}/{}", self.onion, self.shared.local_id)]
+    }
+
+    /// A key, or a location with one (noted as where that key is).
+    fn locate(&self, target: &str) -> Option<String> {
+        if is_key(target) {
+            return Some(target.to_owned());
+        }
+        let (location, key) = parse_location(target)?;
+        let key = key?;
+        self.shared.remember(&key, &location);
+        Some(key)
+    }
+
+    fn locations(&self, target: &str) -> Vec<String> {
+        lock(&self.shared.book)
+            .get(target)
+            .map(|location| format!("{location}/{target}"))
+            .into_iter()
+            .collect()
+    }
+
+    /// A location with a key is that key; one without is dialed, to find out.
+    async fn resolve_seed(&self, seed: &str) -> Result<String> {
+        if is_key(seed) {
+            if lock(&self.shared.book).contains_key(seed) {
+                return Ok(seed.to_owned());
+            }
+            bail!("no onion address known for {seed}; give its onion address");
+        }
+        let (location, key) = parse_location(seed)
+            .ok_or_else(|| anyhow!("{seed} is no onion address with a port"))?;
+        if location == self.onion {
+            bail!("{seed} is this node");
+        }
+        match key {
+            Some(key) => {
+                self.shared.remember(&key, &location);
+                Ok(key)
+            }
+            None => self.dial(&location, None).await,
+        }
+    }
+
+    fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+        let body = wire::encode(message)?;
+        Ok(EncodedMessage {
+            peer: peer.to_owned(),
+            address: peer.to_owned(),
+            payload_len: body.len(),
+            body: body.into(),
+            kind: message.kind(),
+            message: None,
+        })
+    }
+
+    async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> {
+        let key = &message.peer;
+        if !is_key(key) {
+            bail!("{key} is no Tor transport identity");
+        }
+        if message.body.len() > MAX_MESSAGE {
+            bail!(
+                "a message of {} bytes; at most {MAX_MESSAGE} go over Tor",
+                message.body.len()
+            );
+        }
+        let connection = self.connection(key).await?;
+        let timeout = stream_timeout(message.body.len());
+        let written = tokio::time::timeout(timeout, async {
+            let mut writer = connection.writer.lock().await;
+            writer
+                .write_all(&(message.body.len() as u32).to_be_bytes())
+                .await?;
+            writer.write_all(&message.body).await?;
+            writer.flush().await
+        })
+        .await;
+        let result = match written {
+            Ok(Ok(())) => Ok(()),
+            Ok(Err(error)) => Err(anyhow::Error::from(error)),
+            Err(_) => Err(anyhow!("sending to {key} took longer than {timeout:?}")),
+        };
+        if result.is_err() {
+            // Dialed again on the next send.
+            self.shared.forget(key, connection.id);
+        }
+        result.with_context(|| format!("Sending to {key} over Tor"))
+    }
+
+    /// Handles the messages read from every connection, one at a time, in the order
+    /// they were read.
+    async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
+        let Some(mut incoming) = self.incoming.lock().await.take() else {
+            bail!("Tor consumer already started");
+        };
+        while let Some((peer, body)) = incoming.recv().await {
+            match wire::decode(&body) {
+                Ok(message) => {
+                    if let Err(error) = handler(peer.clone(), message, body.len()).await {
+                        warn!(peer = %peer, error = %format!("{error:#}"), "Failed to process inbound message");
+                    }
+                }
+                Err(error) => {
+                    debug!(peer = %peer, error = %error, "Dropping a Tor message that doesn't decode");
+                }
+            }
+        }
+        Ok(())
+    }
+
+    async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+        let mut stopped = self.stopped.clone();
+        let _ = stopped.wait_for(|stopped| *stopped).await;
+        Err(std::io::Error::other("the onion service stopped"))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const ONION: &str = "vww6ybal4bd7szmgncyruucpgfkqahzddi37ktceo3ah7ngmcopnpyyd.onion";
+
+    #[test]
+    fn locations_and_keys() {
+        let key = "ab".repeat(32);
+        assert!(is_key(&key));
+        assert!(!is_key(&key.to_uppercase()));
+        assert!(!is_key(&key[1..]));
+        assert_eq!(
+            parse_location(&format!("{ONION}:9662/{key}")),
+            Some((format!("{ONION}:9662"), Some(key.clone())))
+        );
+        assert_eq!(
+            parse_location(&format!("{ONION}:9662")),
+            Some((format!("{ONION}:9662"), None))
+        );
+        // A port is needed, a key must be one, and only onion addresses.
+        assert_eq!(parse_location(ONION), None);
+        assert_eq!(parse_location(&format!("{ONION}:9662/xyz")), None);
+        assert_eq!(parse_location("example.org:9662"), None);
+        assert_eq!(parse_location(&format!("{ONION}:99999")), None);
+    }
+}