diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/transport/stats.rs | 6 | ||||
| -rw-r--r-- | src/transport/tor.rs | 451 |
2 files changed, 266 insertions, 191 deletions
diff --git a/src/transport/stats.rs b/src/transport/stats.rs index 6b99999..8b456cc 100644 --- a/src/transport/stats.rs +++ b/src/transport/stats.rs @@ -48,11 +48,11 @@ pub(crate) struct PathsStats { #[serde(rename_all = "camelCase")] pub(crate) struct PathInfo { /// `<scheme>:<transport identity>`: whom the path goes to, as the transport knows it - /// (a libp2p peer ID, a Mycelium public key, a node key for Tor). + /// (a libp2p peer ID, a Mycelium public key, an onion service). pub(crate) address: String, /// Where the path goes now (`Transport::location`): for libp2p, the remote address of - /// its connection; for Tor, the onion service. Absent where the address says it all - /// (Mycelium), or with no connection. + /// its connection. Absent where the address says it all (Mycelium, Tor), or with no + /// connection. pub(crate) location: Option<String>, /// Smoothed round-trip time of its PINGs (ours, and our keepalives); absent before /// an answer. diff --git a/src/transport/tor.rs b/src/transport/tor.rs index 91d7f86..b49cc67 100644 --- a/src/transport/tor.rs +++ b/src/transport/tor.rs @@ -3,20 +3,26 @@ //! 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. +//! 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 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. +//! 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 @@ -26,7 +32,7 @@ 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 std::time::{Duration, Instant}; use anyhow::{Context, Result, anyhow, bail}; use arti_client::config::TorClientConfigBuilder; @@ -34,7 +40,7 @@ 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::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; @@ -51,7 +57,7 @@ use crate::transport::{ is_probe, wire, }; -/// A message read, with the sender's key and its payload length. +/// A message read, with the sender's onion service and its payload length. type Inbound = (String, ParsedMessage, usize); /// The onion service's nickname in arti's state (one per node). @@ -69,7 +75,11 @@ const STREAM_TIMEOUT: Duration = Duration::from_secs(30); 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. +/// 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. @@ -88,7 +98,24 @@ pub(crate) struct TorConfig { pub(crate) port: u16, } -/// Whether `id` is a Tor transport identity: a public key, 64 lowercase hex digits. +/// `<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 @@ -96,22 +123,31 @@ fn is_key(id: &str) -> bool { .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; +/// 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); } - port.parse::<u16>().ok()?; - if key.is_some_and(|key| !is_key(key)) { - return 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"), } - 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. @@ -122,12 +158,17 @@ struct Connection { struct Shared { key: Arc<NodeKey>, + client: Arc<TorClient<PreferredRuntime>>, + /// Our onion service, `<onion>.onion:<port>`: our transport identity. 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. + /// 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 location. + /// 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 @@ -142,9 +183,6 @@ fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { 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>, /// Messages, and PINGs and PONGs (`Shared::inbound`, `Shared::probes`). @@ -200,13 +238,14 @@ impl TorTransport { // 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 (probes, incoming_probes) = mpsc::channel(READS); let shared = Arc::new(Shared { key: config.key, - local_id, - book: Mutex::default(), + client, + local_id: onion.clone(), + proven: Mutex::default(), + failed_checks: Mutex::default(), connections: Mutex::default(), dialing: Mutex::default(), next_connection: AtomicU64::new(0), @@ -216,15 +255,13 @@ impl TorTransport { 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 }); + tokio::spawn(async move { shared.accept(request, port).await }); } warn!("The onion service stopped"); let _ = stopping.send(true); @@ -233,95 +270,120 @@ impl TorTransport { info!(address = %onion, "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, incoming_probes))), 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) { +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 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 dialing = self.dialing(location); 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()); + if let Some(connection) = lock(&self.connections).get(location) { + return Ok(Arc::clone(connection)); } - 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, - ) + 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 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) + let mut proven = lock(&self.proven); + if proven.len() < MAX_PEERS || proven.contains_key(location) { + proven.insert(location.to_owned(), key); } - .await; - lock(&self.shared.dialing).remove(location); - result + Ok((key, stream)) } -} -impl Shared { - /// `key` is at `location`. - fn remember(&self, key: &str, location: &str) { - if key == self.local_id { - return; + /// 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()); + } } - let mut book = lock(&self.book); - if book.len() < MAX_PEERS || book.contains_key(key) { - book.insert(key.to_owned(), location.to_owned()); + 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. - async fn accept(self: Arc<Self>, request: StreamRequest, onion: &str, port: u16) { + /// 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 @@ -336,43 +398,70 @@ impl Shared { return; } }; - match challenge::run( - &mut stream, - &self.key, - Role::Listener, - onion, - HANDSHAKE_TIMEOUT, - ) - .await - { - Ok(key) => self.register(Hex(&key).to_string(), stream), + 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 = %location, key = %Hex(&key), "Dropping a connection to our onion service: another node is at the onion service it claims"); + } + Err(error) => { + debug!(claimed = %location, error = %format!("{error:#}"), "Dropping a connection to our onion service: couldn't check the onion service it claims"); } } } - /// 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) { + /// 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(key.clone(), connection); - debug!(peer = %key, connection = id, "Connection opened"); + lock(&self.connections).insert(location.clone(), Arc::clone(&connection)); + debug!(peer = %location, connection = id, "Connection opened"); let shared = Arc::clone(self); tokio::spawn(async move { - let error = shared.read(&key, reader).await; - debug!(peer = %key, connection = id, error = %format!("{error:#}"), "Connection closed"); - shared.forget(&key, id); + let error = shared.read(&location, reader).await; + debug!(peer = %location, connection = id, error = %format!("{error:#}"), "Connection closed"); + shared.forget(&location, id); }); + connection } - /// Reads messages from `key` until the connection fails; returns why. - async fn read(&self, key: &str, mut reader: ReadHalf<DataStream>) -> anyhow::Error { + /// 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 { @@ -396,7 +485,7 @@ impl Shared { let message = match wire::decode(&body) { Ok(message) => message, Err(error) => { - debug!(peer = %key, error = %error, "Dropping a message that doesn't decode"); + debug!(peer = %location, error = %error, "Dropping a message that doesn't decode"); continue; } }; @@ -406,7 +495,7 @@ impl Shared { &self.inbound }; if channel - .send((key.to_owned(), message, body.len())) + .send((location.to_owned(), message, body.len())) .await .is_err() { @@ -415,14 +504,14 @@ impl Shared { } } - /// The connection `id` to `key` is gone: not to be written to any more. - fn forget(&self, key: &str, id: u64) { + /// 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(key) + .get(location) .is_some_and(|connection| connection.id == id) { - connections.remove(key); + connections.remove(location); } } } @@ -434,7 +523,7 @@ impl Transport for TorTransport { } fn is_peer_id(&self, id: &str) -> bool { - is_key(id) + parse_location(id).is_some_and(|location| location == id) } /// The longest a message may take. @@ -442,55 +531,24 @@ impl Transport for TorTransport { stream_timeout(MAX_MESSAGE) } - /// Our onion service, with our key. fn local_addresses(&self) -> Vec<String> { - vec![format!("{}/{}", self.onion, self.shared.local_id)] + vec![self.shared.local_id.clone()] } - /// A key, or a location with one (noted as where that key is). + /// 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> { - 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) - } - - /// The onion service, `<onion>.onion:<port>`. - fn location(&self, target: &str) -> Option<String> { - lock(&self.shared.book).get(target).cloned() + parse_location(target) } - 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. + /// An onion service is its own transport identity: nothing 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) + let location = parse_location(seed) .ok_or_else(|| anyhow!("{seed} is no onion address with a port"))?; - if location == self.onion { + if location == self.shared.local_id { bail!("{seed} is this node"); } - match key { - Some(key) => { - self.shared.remember(&key, &location); - Ok(key) - } - None => self.dial(&location, None).await, - } + Ok(location) } fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { @@ -506,9 +564,9 @@ impl Transport for TorTransport { } async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { - let key = &message.peer; - if !is_key(key) { - bail!("{key} is no Tor transport identity"); + let location = &message.peer; + if !self.is_peer_id(location) { + bail!("{location} is no Tor transport identity"); } if message.body.len() > MAX_MESSAGE { bail!( @@ -516,7 +574,7 @@ impl Transport for TorTransport { message.body.len() ); } - let connection = self.connection(key).await?; + 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; @@ -530,13 +588,15 @@ impl Transport for TorTransport { 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:?}")), + Err(_) => Err(anyhow!( + "sending to {location} took longer than {timeout:?}" + )), }; if result.is_err() { // Dialed again on the next send. - self.shared.forget(key, connection.id); + self.shared.forget(location, connection.id); } - result.with_context(|| format!("Sending to {key} over Tor")) + result.with_context(|| format!("Sending to {location}")) } /// Handles the messages read from every connection, one at a time, in the order @@ -576,18 +636,21 @@ mod tests { const ONION: &str = "vww6ybal4bd7szmgncyruucpgfkqahzddi37ktceo3ah7ngmcopnpyyd.onion"; #[test] - fn locations_and_keys() { + 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((format!("{ONION}:9662"), Some(key.clone()))) + Some(location.clone()) ); assert_eq!( - parse_location(&format!("{ONION}:9662")), - Some((format!("{ONION}:9662"), None)) + 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); @@ -595,4 +658,16 @@ mod tests { 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()); + } } |