path: root/src/transport/tor.rs

//! 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 onion service, `<onion>.onion:<port>`. 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: whoever dials arrives anonymously. So at the start of
//! every connection, the challenge handshake (`transport::challenge`) proves both node
//! keys, bound to the onion address dialed, and then the dialer says which onion
//! service is its own (`send_claim`): the connection's peer from then on. That claim is
//! checked before the connection is used (`Shared::check`): the onion service must be
//! where the dialer's node key is. A dial proves that (Tor proves the onion service, the
//! handshake the key there), so a claim holds if dialing the claimed onion service found
//! the same key, now or before; if it wasn't dialed yet, the listener dials it to check,
//! and closes that connection again. A claim whose check failed is refused without a
//! dial for a while (`FAILED_CHECK_FOR`). 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, Instant};

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::{AsyncRead, AsyncReadExt, AsyncWrite, 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::{
    Addressed, EncodedMessage, HandlerFn, Hex, MessageHandler, ParsedMessage, Transport,
    hand_on_probe, is_probe, wire,
};

/// A message read, with the sender's onion service and its payload length.
type Inbound = (String, ParsedMessage, usize);

/// The scheme of this transport's addresses (`transport::paths`).
pub(crate) const SCHEME: &str = "tor";

/// 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;
/// A claim whose check failed (`Shared::check`) is refused without dialing again for
/// this long: claims of onion services that aren't the claimer's don't make us dial
/// over and over. A dial of our own there that finds the claimer's key ends it early.
const FAILED_CHECK_FOR: Duration = Duration::from_secs(600);
/// Onion services remembered (with the node key found there, `Shared::proven`), 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,
}

/// `<onion>.onion:<port>`, from `target`, which may end in `/<node key>` (how nodes
/// published their onion service before it was their transport identity).
fn parse_location(target: &str) -> Option<String> {
    let location = match target.split_once('/') {
        Some((location, key)) if is_key(key) => location,
        Some(_) => return None,
        None => target,
    };
    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()?;
    Some(location.to_ascii_lowercase())
}

/// Whether `id` is a node 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'))
}

/// After the handshake, the dialer says what the connection is for (`read_claim`):
/// carrying messages from its onion service, `claim`, or (empty) only checking which
/// node key is at the listener's (`Shared::check`).
async fn send_claim<S: AsyncWrite + Unpin>(stream: &mut S, claim: &str) -> Result<()> {
    let length = u8::try_from(claim.len()).map_err(|_| anyhow!("{claim} is too long"))?;
    stream.write_all(&[length]).await?;
    stream.write_all(claim.as_bytes()).await?;
    stream.flush().await?;
    Ok(())
}

/// The dialer's claim (`send_claim`): its onion service, or None for a check.
async fn read_claim<S: AsyncRead + Unpin>(stream: &mut S) -> Result<Option<String>> {
    let mut length = [0u8; 1];
    stream.read_exact(&mut length).await?;
    if length[0] == 0 {
        return Ok(None);
    }
    let mut claim = vec![0u8; length[0] as usize];
    stream.read_exact(&mut claim).await?;
    let claim = String::from_utf8(claim).map_err(|_| anyhow!("a claim that isn't text"))?;
    match parse_location(&claim) {
        Some(location) if location == claim => Ok(Some(location)),
        _ => bail!("{claim} is no onion address with a port"),
    }
}

/// 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>,
    client: Arc<TorClient<PreferredRuntime>>,
    /// Our onion service, `<onion>.onion:<port>`: our transport identity.
    local_id: String,
    /// The node key found at each onion service dialed: proven there (`check`).
    proven: Mutex<HashMap<String, [u8; 32]>>,
    /// Claims whose check failed lately (`FAILED_CHECK_FOR`): an onion service and the
    /// node key claiming it, and when.
    failed_checks: Mutex<HashMap<(String, [u8; 32]), Instant>>,
    /// The connection to write to each peer on, by onion service.
    connections: Mutex<HashMap<String, Arc<Connection>>>,
    /// One dial at a time per onion service.
    dialing: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
    next_connection: AtomicU64,
    /// Messages read, for the consumer: in order, and PINGs and PONGs apart, to be
    /// handed on at once (`is_probe`).
    inbound: mpsc::Sender<Inbound>,
    probes: mpsc::Sender<Inbound>,
}

fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    mutex.lock_or_recover()
}

pub(crate) struct TorTransport {
    shared: Arc<Shared>,
    /// Kept: dropping it stops the onion service.
    _service: Arc<RunningOnionService>,
    /// Messages, and PINGs and PONGs (`Shared::inbound`, `Shared::probes`).
    incoming: tokio::sync::Mutex<Option<(mpsc::Receiver<Inbound>, mpsc::Receiver<Inbound>)>>,
    stopped: watch::Receiver<bool>,
}

impl TorTransport {
    /// Launches our onion service, and bootstraps arti in the background (that takes a
    /// while: it fetches what it needs to know of the Tor network). Until it is done,
    /// dials wait for it (arti bootstraps on demand), and the onion service is published
    /// once it is: the node starts without waiting for Tor.
    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")?;
        let client = TorClient::builder()
            .config(tor_config)
            .create_unbootstrapped_async()
            .await
            .context("Starting Tor")?;
        {
            let client = client.clone();
            let state_dir = config.state_dir.display().to_string();
            info!(state_dir = %state_dir, "Bootstrapping Tor");
            // Launching the onion service starts it too; this waits for it to be done.
            tokio::spawn(async move {
                match client.bootstrap().await {
                    Ok(()) => info!("Tor bootstrapped"),
                    // Dials try again (on demand).
                    Err(error) => warn!(error = %error, "Bootstrapping Tor failed"),
                }
            });
        }
        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 (inbound, incoming) = mpsc::channel(READS);
        let (probes, incoming_probes) = mpsc::channel(READS);
        let shared = Arc::new(Shared {
            key: config.key,
            client,
            local_id: onion.clone(),
            proven: Mutex::default(),
            failed_checks: Mutex::default(),
            connections: Mutex::default(),
            dialing: Mutex::default(),
            next_connection: AtomicU64::new(0),
            inbound,
            probes,
        });
        let (stopping, stopped) = watch::channel(false);
        {
            let shared = Arc::clone(&shared);
            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);
                    tokio::spawn(async move { shared.accept(request, port).await });
                }
                warn!("The onion service stopped");
                let _ = stopping.send(true);
            });
        }
        info!(address = %onion, "Transport started (the onion service may take a minute to be reachable)");
        Ok(Self {
            shared,
            _service: service,
            incoming: tokio::sync::Mutex::new(Some((incoming, incoming_probes))),
            stopped,
        })
    }
}

impl Shared {
    /// A connection to the onion service at `location`, made if there is none.
    async fn connection(self: &Arc<Self>, location: &str) -> Result<Arc<Connection>> {
        if let Some(connection) = lock(&self.connections).get(location) {
            return Ok(Arc::clone(connection));
        }
        let dialing = self.dialing(location);
        let _one_at_a_time = dialing.lock().await;
        // Someone else may have connected meanwhile.
        if let Some(connection) = lock(&self.connections).get(location) {
            return Ok(Arc::clone(connection));
        }
        let (_, stream) = self.dial(location, &self.local_id).await?;
        Ok(self.register(location.to_owned(), stream))
    }

    /// The lock that keeps dials to `location` one at a time.
    fn dialing(&self, location: &str) -> Arc<tokio::sync::Mutex<()>> {
        let mut dialing = lock(&self.dialing);
        if dialing.len() >= MAX_PEERS {
            // Only those in use.
            dialing.retain(|_, lock| Arc::strong_count(lock) > 1);
        }
        Arc::clone(dialing.entry(location.to_owned()).or_default())
    }

    /// Dials the onion service at `location`, runs the handshake and says what for
    /// (`send_claim`): returns the node key there, now proven there, and the stream.
    async fn dial(&self, location: &str, claim: &str) -> Result<([u8; 32], DataStream)> {
        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.key,
            Role::Dialer,
            location,
            HANDSHAKE_TIMEOUT,
        )
        .await
        .with_context(|| format!("Handshake with {location}"))?;
        tokio::time::timeout(HANDSHAKE_TIMEOUT, send_claim(&mut stream, claim))
            .await
            .map_err(|_| anyhow!("no handshake with {location} within {HANDSHAKE_TIMEOUT:?}"))?
            .with_context(|| format!("Handshake with {location}"))?;
        let mut proven = lock(&self.proven);
        if proven.len() < MAX_PEERS || proven.contains_key(location) {
            proven.insert(location.to_owned(), key);
        }
        Ok((key, stream))
    }

    /// Whether the node with `key` is at the onion service at `location` (a dialer
    /// claims it is: module docs). Dialed to find out, unless that was done already, or
    /// failed within `FAILED_CHECK_FOR`.
    async fn check(&self, location: &str, key: [u8; 32]) -> Result<bool> {
        if lock(&self.proven).get(location) == Some(&key) {
            return Ok(true);
        }
        let claim = (location.to_owned(), key);
        if let Some(failed) = self.failed_lately(&claim) {
            bail!("its check failed {}s ago", failed.as_secs());
        }
        let dialing = self.dialing(location);
        let _one_at_a_time = dialing.lock().await;
        // A dial or check meanwhile may have found out.
        if lock(&self.proven).get(location) == Some(&key) {
            return Ok(true);
        }
        if let Some(failed) = self.failed_lately(&claim) {
            bail!("its check failed {}s ago", failed.as_secs());
        }
        let checked = self
            .dial(location, "")
            .await
            .map(|(found, _check_only)| found == key);
        if !matches!(checked, Ok(true)) {
            let mut failed = lock(&self.failed_checks);
            if failed.len() >= MAX_PEERS {
                failed.retain(|_, at| at.elapsed() < FAILED_CHECK_FOR);
            }
            if failed.len() < MAX_PEERS {
                failed.insert(claim, Instant::now());
            }
        }
        checked
    }

    /// How long ago the check of `claim` failed, if within `FAILED_CHECK_FOR`.
    fn failed_lately(&self, claim: &(String, [u8; 32])) -> Option<Duration> {
        let mut failed = lock(&self.failed_checks);
        let since = failed.get(claim)?.elapsed();
        if since < FAILED_CHECK_FOR {
            return Some(since);
        }
        failed.remove(claim);
        None
    }

    /// A stream to our onion service: accepted if for our port, then the handshake, and
    /// the dialer's claim checked (module docs).
    async fn accept(self: Arc<Self>, request: StreamRequest, 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;
            }
        };
        let handshake = async {
            let key = challenge::run(
                &mut stream,
                &self.key,
                Role::Listener,
                &self.local_id,
                HANDSHAKE_TIMEOUT,
            )
            .await?;
            let claim = tokio::time::timeout(HANDSHAKE_TIMEOUT, read_claim(&mut stream))
                .await
                .map_err(|_| anyhow!("no claim within {HANDSHAKE_TIMEOUT:?}"))??;
            Ok::<_, anyhow::Error>((key, claim))
        };
        let (key, claim) = match handshake.await {
            Ok(done) => done,
            Err(error) => {
                debug!(error = %format!("{error:#}"), "Dropping a connection to our onion service: the handshake failed");
                return;
            }
        };
        // A check (`check`): the handshake was all.
        let Some(location) = claim else {
            return;
        };
        if location == self.local_id {
            debug!("Dropping a connection to our onion service: it claims to be ours");
            return;
        }
        match self.check(&location, key).await {
            Ok(true) => {
                self.register(location, stream);
            }
            Ok(false) => {
                debug!(claimed = %Addressed(SCHEME, &location), key = %Hex(&key), "Dropping a connection to our onion service: another node is at the onion service it claims");
            }
            Err(error) => {
                debug!(claimed = %Addressed(SCHEME, &location), error = %format!("{error:#}"), "Dropping a connection to our onion service: couldn't check the onion service it claims");
            }
        }
    }

    /// A connection to the onion service at `location`, authenticated: the one to write
    /// to it on from now, and read from until it closes.
    fn register(self: &Arc<Self>, location: String, stream: DataStream) -> Arc<Connection> {
        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(location.clone(), Arc::clone(&connection));
        debug!(peer = %Addressed(SCHEME, &location), connection = id, "Connection opened");
        let shared = Arc::clone(self);
        tokio::spawn(async move {
            let error = shared.read(&location, reader).await;
            debug!(peer = %Addressed(SCHEME, &location), connection = id, error = %format!("{error:#}"), "Connection closed");
            shared.forget(&location, id);
        });
        connection
    }

    /// Reads messages from `location` until the connection fails; returns why.
    async fn read(&self, location: &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)
                    );
                }
            }
            let message = match wire::decode(&body) {
                Ok(message) => message,
                Err(error) => {
                    debug!(peer = %Addressed(SCHEME, &location), error = %error, "Dropping a message that doesn't decode");
                    continue;
                }
            };
            let channel = if is_probe(&message) {
                &self.probes
            } else {
                &self.inbound
            };
            if channel
                .send((location.to_owned(), message, body.len()))
                .await
                .is_err()
            {
                return anyhow!("the transport stopped");
            }
        }
    }

    /// The connection `id` to `location` is gone: not to be written to any more.
    fn forget(&self, location: &str, id: u64) {
        let mut connections = lock(&self.connections);
        if connections
            .get(location)
            .is_some_and(|connection| connection.id == id)
        {
            connections.remove(location);
        }
    }
}

#[async_trait]
impl Transport for TorTransport {
    fn local_id(&self) -> &str {
        &self.shared.local_id
    }

    fn is_peer_id(&self, id: &str) -> bool {
        parse_location(id).is_some_and(|location| location == id)
    }

    /// The longest a message may take.
    fn delivery_limit(&self) -> Duration {
        stream_timeout(MAX_MESSAGE)
    }

    fn local_addresses(&self) -> Vec<String> {
        vec![self.shared.local_id.clone()]
    }

    /// An onion service (`<onion>.onion:<port>`, with a node key after it as nodes
    /// published it before, which is left out).
    fn locate(&self, target: &str) -> Option<String> {
        parse_location(target)
    }

    /// An onion service is its own transport identity: nothing to find out.
    async fn resolve_seed(&self, seed: &str) -> Result<String> {
        let location = parse_location(seed)
            .ok_or_else(|| anyhow!("{seed} is no onion address with a port"))?;
        if location == self.shared.local_id {
            bail!("{seed} is this node");
        }
        Ok(location)
    }

    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 location = &message.peer;
        if !self.is_peer_id(location) {
            bail!("{location} 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.shared.connection(location).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 {location} took longer than {timeout:?}"
            )),
        };
        if result.is_err() {
            // Dialed again on the next send.
            self.shared.forget(location, connection.id);
        }
        result.with_context(|| format!("Sending to {location}"))
    }

    /// 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, mut probes)) = self.incoming.lock().await.take() else {
            bail!("Tor consumer already started");
        };
        let handler: Arc<HandlerFn> = Arc::from(handler);
        {
            let handler = Arc::clone(&handler);
            tokio::spawn(async move {
                while let Some((peer, message, payload_len)) = probes.recv().await {
                    hand_on_probe(SCHEME, &handler, peer, message, payload_len);
                }
            });
        }
        while let Some((peer, message, payload_len)) = incoming.recv().await {
            if let Err(error) = handler(peer.clone(), message, payload_len).await {
                warn!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Failed to process inbound message");
            }
        }
        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() {
        let key = "ab".repeat(32);
        assert!(is_key(&key));
        assert!(!is_key(&key.to_uppercase()));
        assert!(!is_key(&key[1..]));
        let location = format!("{ONION}:9662");
        assert_eq!(parse_location(&location), Some(location.clone()));
        // As nodes published them before: the key is left out.
        assert_eq!(
            parse_location(&format!("{ONION}:9662/{key}")),
            Some(location.clone())
        );
        assert_eq!(
            parse_location(&location.to_uppercase().replace(".ONION", ".onion")),
            Some(location)
        );
        // 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);
    }

    #[tokio::test]
    async fn claims_round_trip() {
        let (mut dialer, mut listener) = tokio::io::duplex(1024);
        let location = format!("{ONION}:9662");
        send_claim(&mut dialer, &location).await.unwrap();
        send_claim(&mut dialer, "").await.unwrap();
        send_claim(&mut dialer, "example.org:1").await.unwrap();
        assert_eq!(read_claim(&mut listener).await.unwrap(), Some(location));
        assert_eq!(read_claim(&mut listener).await.unwrap(), None);
        assert!(read_claim(&mut listener).await.is_err());
    }
}