diff options
Diffstat (limited to 'src/transport/tor.rs')
| -rw-r--r-- | src/transport/tor.rs | 673 |
1 files changed, 673 insertions, 0 deletions
diff --git a/src/transport/tor.rs b/src/transport/tor.rs new file mode 100644 index 0000000..b49cc67 --- /dev/null +++ b/src/transport/tor.rs @@ -0,0 +1,673 @@ +//! 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::{ + 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 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 = %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 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 = %location, connection = id, "Connection opened"); + let shared = Arc::clone(self); + tokio::spawn(async move { + 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 `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 = %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(&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 = %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()); + } +} |