Add M0 Arti two-node onion byte-pipe test.

Two in-process TorClients host v3 onions, echo ping over Tor, keep the
same locator across restart, and still accept after DormantMode::Soft.

Cargo.toml

@@ -7,3 +7,14 @@ license = "MIT"
 publish = false
 
 [dependencies]
+arti-client = { version = "0.46", features = ["tokio", "onion-service-client", "onion-service-service"] }
+futures = "0.3"
+safelog = "0.9"
+tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "time"] }
+tor-cell = "0.46"
+tor-hsservice = "0.46"
+tor-rtcompat = { version = "0.46", features = ["tokio"] }
+
+[dev-dependencies]
+tempfile = "3"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }

tests/tor_hs.rs

@@ -0,0 +1,223 @@
+//! M0: two in-process Arti clients, each hosting a v3 onion, echo ping.
+//!
+//! Needs a live Tor network. `cargo test --test tor_hs` skips it;
+//! run with `--ignored` on a networked machine.
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use arti_client::config::TorClientConfigBuilder;
+use arti_client::{DormantMode, TorClient, TorClientConfig};
+use futures::StreamExt;
+use futures::io::{AsyncReadExt, AsyncWriteExt};
+use safelog::DisplayRedacted;
+use tor_cell::relaycell::msg::Connected;
+use tor_hsservice::status::State;
+use tor_hsservice::{HsNickname, OnionServiceConfig, RunningOnionService, handle_rend_requests};
+use tor_rtcompat::PreferredRuntime;
+
+const PING: &[u8] = b"ping";
+const HS_PORT: u16 = 80;
+const BOOTSTRAP_LOG: &str = "info";
+
+type Client = Arc<TorClient<PreferredRuntime>>;
+
+fn client_config(state_dir: &std::path::Path, cache_dir: &std::path::Path) -> TorClientConfig {
+    std::fs::create_dir_all(state_dir).expect("state dir");
+    std::fs::create_dir_all(cache_dir).expect("cache dir");
+    let mut builder = TorClientConfigBuilder::from_directories(state_dir, cache_dir);
+    // Temp dirs sit under $TMP; skip fs-mistrust on the parent tree.
+    builder.storage().permissions().dangerously_trust_everyone();
+    builder.build().expect("TorClientConfig")
+}
+
+async fn bootstrapped(state_dir: &std::path::Path, cache_dir: &std::path::Path) -> Client {
+    let cfg = client_config(state_dir, cache_dir);
+    TorClient::create_bootstrapped(cfg)
+        .await
+        .expect("Arti bootstrap failed — fail closed, no C-tor fallback")
+}
+
+fn hs_config(nickname: &str) -> OnionServiceConfig {
+    let nickname = HsNickname::new(nickname.to_string()).expect("HsNickname");
+    OnionServiceConfig::builder()
+        .nickname(nickname)
+        .build()
+        .expect("OnionServiceConfig")
+}
+
+fn onion_string(svc: &RunningOnionService) -> String {
+    let id = svc
+        .onion_address()
+        .expect("onion identity missing from keystore");
+    id.display_unredacted().to_string()
+}
+
+fn spawn_echo(
+    rend: impl futures::Stream<Item = tor_hsservice::RendRequest> + Send + 'static,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        let mut requests = std::pin::pin!(handle_rend_requests(rend));
+        while let Some(req) = requests.next().await {
+            tokio::spawn(async move {
+                let Ok(mut stream) = req.accept(Connected::new_empty()).await else {
+                    return;
+                };
+                let mut buf = [0u8; 64];
+                let Ok(n) = stream.read(&mut buf).await else {
+                    return;
+                };
+                let _ = stream.write_all(&buf[..n]).await;
+                let _ = stream.flush().await;
+            });
+        }
+    })
+}
+
+async fn launch_echo(
+    client: &Client,
+    nickname: &str,
+) -> (Arc<RunningOnionService>, tokio::task::JoinHandle<()>, String) {
+    let launched = client
+        .launch_onion_service(hs_config(nickname))
+        .expect("launch_onion_service")
+        .expect("onion service disabled in config — fail closed");
+    let (svc, rend) = launched;
+    let onion = onion_string(&svc);
+    let echo = spawn_echo(rend);
+    (svc, echo, onion)
+}
+
+async fn wait_until_published(svc: &RunningOnionService, label: &str) {
+    let deadline = Instant::now() + Duration::from_secs(180);
+    let mut events = svc.status_events();
+    loop {
+        let st = svc.status();
+        eprintln!("{label} hs status: {:?}", st.state());
+        match st.state() {
+            State::Running | State::DegradedReachable => return,
+            State::Broken => {
+                panic!("{label}: onion service broken: {:?}", st.current_problem())
+            }
+            _ => {}
+        }
+        if Instant::now() >= deadline {
+            panic!(
+                "{label}: onion service did not publish within 180s: {:?}",
+                st.state()
+            );
+        }
+        tokio::select! {
+            _ = events.next() => {}
+            _ = tokio::time::sleep(Duration::from_secs(2)) => {}
+        }
+    }
+}
+
+async fn echo_ping(client: &Client, onion: &str) -> Result<(), String> {
+    let mut stream = client
+        .connect((onion, HS_PORT))
+        .await
+        .map_err(|e| format!("connect {onion}:{HS_PORT}: {e}"))?;
+    stream
+        .write_all(PING)
+        .await
+        .map_err(|e| format!("write ping: {e}"))?;
+    stream.flush().await.map_err(|e| format!("flush ping: {e}"))?;
+    let mut buf = vec![0u8; PING.len()];
+    stream
+        .read_exact(&mut buf)
+        .await
+        .map_err(|e| format!("read echo: {e}"))?;
+    if buf != PING {
+        return Err(format!("echo payload {buf:?}"));
+    }
+    Ok(())
+}
+
+async fn echo_ping_retry(client: &Client, onion: &str, timeout: Duration) {
+    let deadline = Instant::now() + timeout;
+    let mut last = None::<String>;
+    while Instant::now() < deadline {
+        match tokio::time::timeout(Duration::from_secs(45), echo_ping(client, onion)).await {
+            Ok(Ok(())) => return,
+            Ok(Err(e)) => last = Some(e),
+            Err(_) => last = Some("attempt timeout".to_string()),
+        }
+        tokio::time::sleep(Duration::from_secs(3)).await;
+    }
+    panic!(
+        "echo ping to {onion} failed within {:?}: {:?}",
+        timeout, last
+    );
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "needs live Tor network"]
+async fn two_node_byte_pipe_restart_and_dormant() {
+    let _ = tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| BOOTSTRAP_LOG.into()),
+        )
+        .with_test_writer()
+        .try_init();
+
+    let root = tempfile::tempdir().expect("tempdir");
+    let alice_state = root.path().join("alice/state");
+    let alice_cache = root.path().join("alice/cache");
+    let bob_state = root.path().join("bob/state");
+    let bob_cache = root.path().join("bob/cache");
+
+    eprintln!("bootstrapping alice + bob TorClients…");
+    let (alice, bob) = tokio::join!(
+        bootstrapped(&alice_state, &alice_cache),
+        bootstrapped(&bob_state, &bob_cache),
+    );
+
+    let (alice_svc, alice_echo, alice_onion) = launch_echo(&alice, "onionwire-a").await;
+    let (bob_svc, _bob_echo, bob_onion) = launch_echo(&bob, "onionwire-b").await;
+    eprintln!("alice onion={alice_onion}");
+    eprintln!("bob onion={bob_onion}");
+    assert_ne!(alice_onion, bob_onion, "separate HS identities");
+
+    wait_until_published(&alice_svc, "alice").await;
+    wait_until_published(&bob_svc, "bob").await;
+
+    eprintln!("echo ping alice → bob");
+    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(180)).await;
+
+    // Restart alice: same state dir, same nickname → same onion locator.
+    // Abort the echo task and drop the client so Arti releases the state lock.
+    alice_echo.abort();
+    drop(alice_svc);
+    drop(alice);
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    eprintln!("restarting alice from persisted state…");
+    let alice = bootstrapped(&alice_state, &alice_cache).await;
+    let (alice_svc, _alice_echo, alice_onion_after) = launch_echo(&alice, "onionwire-a").await;
+    assert_eq!(
+        alice_onion, alice_onion_after,
+        "onion locator must be stable across restart"
+    );
+    // Alice as client → Bob does not need Alice's HS republished; the locator
+    // assert above is the restart invariant.
+    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(180)).await;
+
+    // DormantMode::Soft then wake: still accepts within 120s.
+    eprintln!("bob DormantMode::Soft");
+    bob.set_dormant(DormantMode::Soft);
+    tokio::time::sleep(Duration::from_secs(3)).await;
+    eprintln!("bob DormantMode::Normal (wake)");
+    bob.set_dormant(DormantMode::Normal);
+    let t0 = Instant::now();
+    echo_ping_retry(&alice, &bob_onion, Duration::from_secs(120)).await;
+    let woke = t0.elapsed();
+    eprintln!("post-dormant echo succeeded in {woke:?}");
+    assert!(
+        woke <= Duration::from_secs(120),
+        "wake accept took {woke:?} (>120s)"
+    );
+
+    let _ = (alice_svc, bob_svc);
+}