diff --git a/.agents/skills/overlay-functional/SKILL.md b/.agents/skills/overlay-functional/SKILL.md new file mode 100644 index 000000000..83186679b --- /dev/null +++ b/.agents/skills/overlay-functional/SKILL.md @@ -0,0 +1,23 @@ +--- +name: overlay-functional +description: >- + Run or fix the private-mesh overlay harness (Tor chutney-equivalent, + i2pd netid, cjdns TUN). Use when a PR is labeled overlay-functional, a + ship version-bump needs that job, or scripts/overlay-functional is red. + Do not treat this harness as the default PR pin. +--- + +# Overlay functional harness + +Owner: [`docs/overlay-functional.md`](../../../docs/overlay-functional.md). +Default `cargo test` is the PR pin ([`TESTING.md`](../../../TESTING.md)). +Fake SOCKS / control / SAM stay in unlabeled product PRs. Label this +harness and every ship version-bump PR. + +When the labeled job is red, reproduce locally. Do not push and wait on +GitHub Actions as the inner loop. + +```bash +export CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-$PWD/target/dev}" +./scripts/overlay-functional/run.sh +``` diff --git a/.agents/skills/ship-pr/SKILL.md b/.agents/skills/ship-pr/SKILL.md index f502c7973..346cb27a8 100644 --- a/.agents/skills/ship-pr/SKILL.md +++ b/.agents/skills/ship-pr/SKILL.md @@ -82,6 +82,7 @@ Suite and budgets: [`TESTING.md`](../../../TESTING.md). | After Green | `cargo test --workspace --quiet` | | After Refactor, before commit | Local CI except coverage (below) | | Core functional CI red | [core-functional skill](../core-functional/SKILL.md) | +| Overlay functional CI red | [overlay-functional skill](../overlay-functional/SKILL.md) | | Never local | `./scripts/coverage.sh`, `nix build .#rbitcoin-musl`, host IBD | Redirect every command above (stdout and stderr to `/tmp`, print `EXIT`, @@ -120,6 +121,9 @@ not operator zips. Label `core-functional` when the PR touches that harness, and on every ship version-bump PR. Do not label ordinary net or RPC PRs. +Label `overlay-functional` when the PR touches that harness, and on every +ship version-bump PR. Poll with `--interest overlay-functional`. It is not +a required check. Label `nixos-module-runtime` when the NixOS module VM test should run (not eval). Poll with `--interest nixos-module-runtime`. It is not a required check. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e93275b8e..dc2513499 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -127,6 +127,8 @@ jobs: run: ./scripts/ci-os-smoke.test.sh - name: pr-checks-watch self-test run: ./scripts/pr-checks-watch.test.sh + - name: overlay-functional runner self-test + run: ./scripts/overlay-functional/run.sh.test.sh - name: cargo test run: cargo test --workspace diff --git a/.github/workflows/overlay-functional.yml b/.github/workflows/overlay-functional.yml new file mode 100644 index 000000000..202fc1f3f --- /dev/null +++ b/.github/workflows/overlay-functional.yml @@ -0,0 +1,82 @@ +# Overlay functional gate: nightly + dispatch + ship PRs + labels. +# Unlabeled non-ship PRs do not start the heavy job. Default cargo test +# never calls run.sh. +name: overlay-functional + +on: + schedule: + - cron: "42 6 * * *" + workflow_dispatch: + pull_request: + types: [opened, synchronize, reopened, labeled, unlabeled] + +concurrency: + group: overlay-functional-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + detect: + name: detect-ship + runs-on: ubuntu-latest + outputs: + ship: ${{ steps.v.outputs.ship }} + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + if: github.event_name == 'pull_request' + with: + persist-credentials: false + - name: Classify ship vs .99 + id: v + run: | + set -euo pipefail + if [[ "${{ github.event_name }}" != "pull_request" ]]; then + echo "ship=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + kind="$(./scripts/release-gate.sh --kind)" + echo "kind=$kind" >> "$GITHUB_OUTPUT" + if [[ "$kind" == ship ]]; then + echo "ship=true" >> "$GITHUB_OUTPUT" + else + echo "ship=false" >> "$GITHUB_OUTPUT" + fi + + overlay-functional: + name: overlay-functional + needs: [detect] + runs-on: ubuntu-latest + timeout-minutes: 45 + if: >- + always() && + needs.detect.result == 'success' && + (github.event_name != 'pull_request' || + needs.detect.outputs.ship == 'true' || + contains(github.event.pull_request.labels.*.name, 'overlay-functional') || + contains(github.event.pull_request.labels.*.name, 'release')) + env: + CARGO_TARGET_DIR: target/dev + RUSTFLAGS: -D warnings + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: cachix/install-nix-action@13d8dd58da0234aa297dedd986986ccb8e7f3e24 # v31.11.1 + with: + github_access_token: ${{ github.token }} + enable_kvm: false + - uses: nix-community/cache-nix-action@7df957e333c1e5da7721f60227dbba6d06080569 # v7 + with: + primary-key: overlay-functional-${{ runner.os }}-${{ hashFiles('flake.lock') }} + restore-prefixes-first-match: overlay-functional-${{ runner.os }}- + gc-max-store-size-linux: 4G + - uses: Swatinem/rust-cache@6323deb102c322ba6fcbdcafc7e3dddab59af2b6 # v2.9.2 + with: + shared-key: overlay-functional + workspaces: ". -> target/dev" + - name: TUN + run: sudo modprobe tun + - name: overlay journeys + run: nix develop .#overlayFunctional --command ./scripts/overlay-functional/run.sh diff --git a/OPERATOR.md b/OPERATOR.md index dbfe07987..f73267131 100644 --- a/OPERATOR.md +++ b/OPERATOR.md @@ -488,7 +488,10 @@ Omit ADDR for `127.0.0.1:7656`. Failed HELLO / `SESSION CREATE` is a start error. Unset: I2P rows may still load from `peers` v2 but are not dialed. `--only-net i2p` without `--i2p-sam` is a start error. `--i2p-accept-incoming` creates a persistent local destination (`{datadir}/i2p/p2p.priv`, 0600) and -`STREAM FORWARD`s to the P2P bind. With `--listen=0` that is a start error +`STREAM FORWARD`s to the P2P bind. That destination is published as +`{52}.b32.i2p:0` (SAM 3.1 has no ports; Core refuses any other I2P port) +on `getnetworkinfo.localaddresses` and in addrv2, including with +`--no-discover`. With `--listen=0` that is a start error unless `--listen-onion` provides a loopback accept. NixOS: `services.rbitcoin.i2p.sam` / `i2p.acceptIncoming`; the unit `After`/`Wants` `i2pd.service` when SAM is set. Do not start i2pd from this module. diff --git a/TESTING.md b/TESTING.md index 80f5c6c80..ee487dce2 100644 --- a/TESTING.md +++ b/TESTING.md @@ -18,6 +18,10 @@ Electrum TCP, Esplora HTTP, BIP324 P2P). Bitcoin Core’s Python functional suite is a **nightly / ship / `core-functional` label** oracle. It is **not** a required PR check and **not** required on PRs that merely touch net or RPC (too slow). Nightly Core is not a license to delete in-tree tests. +Unlabeled PRs stay cargo-only for overlays too: live Tor / i2pd / cjdns +meshes are the **`overlay-functional`** labeled / nightly / ship job +([`docs/overlay-functional.md`](docs/overlay-functional.md)), not default +`cargo test`. When adding or folding a pin: @@ -497,6 +501,10 @@ suite. A red labeled run is reproduced locally with `./scripts/core-functional/run.sh ` until that script passes ([`docs/core-functional.md`](docs/core-functional.md)); do not push-and-wait on that job as the inner loop. +Label **`overlay-functional`** on harness PRs and on **ship** version-bump +PRs. It is **not** a required PR check. A red labeled run is reproduced +locally with `./scripts/overlay-functional/run.sh ` +([`docs/overlay-functional.md`](docs/overlay-functional.md)). ```bash python3 scripts/core-functional/check_inventory.py @@ -511,4 +519,6 @@ python3 scripts/core-functional/check_inventory.py # cargo test stages Core JSON from the submodule: ./scripts/core-functional/init-submodule.sh ./scripts/core-functional/sync-core-fixtures.sh --check +./scripts/overlay-functional/run.sh.test.sh +./scripts/overlay-functional/run.sh --list ``` diff --git a/crates/rbitcoin-net/src/ephemeral.rs b/crates/rbitcoin-net/src/ephemeral.rs index 258057c38..78ae0b98d 100644 --- a/crates/rbitcoin-net/src/ephemeral.rs +++ b/crates/rbitcoin-net/src/ephemeral.rs @@ -121,7 +121,15 @@ pub fn spawn_isolated_broadcast_loop( Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }; - let Some(tx) = mp.get_tx(&txid) else { + // `get_tx` refuses the reactor thread. This task is a tokio worker. + let tx = match mp.try_get_tx(&txid) { + Some(tx) => Some(tx), + None => { + tokio::time::sleep(Duration::from_millis(10)).await; + mp.try_get_tx(&txid) + } + }; + let Some(tx) = tx else { continue; }; isolated_broadcast_known_tx(&dialer, &addrman, magic, &user_agent, txid, tx).await; @@ -414,6 +422,30 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn isolated_broadcast_loop_reads_mempool_off_reactor() { + let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("iso-reactor"); + let mp = MempoolHub::open(dir.join("mp"), Arc::clone(&hub.query)).unwrap(); + mp.set_isolated_broadcast(true); + let am = Arc::new(Mutex::new(AddrMan::new())); + let h = spawn_isolated_broadcast_loop( + mp.clone(), + Dialer::Direct, + am, + Magic::REGTEST, + "/rbitcoin:test/".into(), + ); + tokio::time::sleep(Duration::from_millis(20)).await; + mp.mark_local_origin(dummy_tx().compute_txid()); + tokio::time::sleep(Duration::from_millis(80)).await; + assert!( + !h.is_finished(), + "broadcast task must not panic reading the mempool on a tokio worker" + ); + h.abort(); + let _ = std::fs::remove_dir_all(dir); + } + #[tokio::test] async fn ephemeral_broadcast_loop_skips_unknown_txid() { let (dir, hub) = crate::chain::tiny_regtest_hub_labeled("iso-skip"); diff --git a/crates/rbitcoin-net/src/i2p_sam.rs b/crates/rbitcoin-net/src/i2p_sam.rs index a30489a8c..4efd43bb7 100644 --- a/crates/rbitcoin-net/src/i2p_sam.rs +++ b/crates/rbitcoin-net/src/i2p_sam.rs @@ -1,11 +1,12 @@ //! I2P SAM v3 STREAM CONNECT / FORWARD (system router, not SOCKS). use crate::error::NetError; +use bitcoin::hashes::{sha256, Hash}; use std::io::Write; use std::net::SocketAddr; use std::path::Path; use std::sync::Mutex; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; static INSTALLED: Mutex> = Mutex::new(None); @@ -13,12 +14,14 @@ static INSTALLED: Mutex> = Mutex::new(None); struct Installed { dialer: I2pDialer, _keepalive: Option, + _forward: Option, } pub struct I2pSam { sam_addr: SocketAddr, session_id: String, destination: String, + forward_port: Option, _control: TcpStream, _forward: Option, } @@ -28,19 +31,22 @@ pub struct I2pDialer { sam_addr: SocketAddr, session_id: String, destination: String, + forward_port: Option, } pub fn install(dialer: I2pDialer) { *INSTALLED.lock().unwrap_or_else(|e| e.into_inner()) = Some(Installed { dialer, _keepalive: None, + _forward: None, }); } -fn install_kept(dialer: I2pDialer, keepalive: TcpStream) { +fn install_kept(dialer: I2pDialer, keepalive: TcpStream, forward: Option) { *INSTALLED.lock().unwrap_or_else(|e| e.into_inner()) = Some(Installed { dialer, _keepalive: Some(keepalive), + _forward: forward, }); } @@ -49,8 +55,8 @@ pub async fn stream_connect_installed(dest_b32: &str) -> Result Ok(s), Err(e) if session_dead(&e) => { - let (fresh, keepalive) = first.recreate_session().await?; - install_kept(fresh.clone(), keepalive); + let (fresh, keepalive, forward) = first.recreate_session().await?; + install_kept(fresh.clone(), keepalive, forward); fresh.stream_connect(dest_b32).await } Err(e) => Err(e), @@ -79,6 +85,39 @@ fn session_dead(err: &NetError) -> bool { } } +fn sam_retry(err: &NetError) -> bool { + match err { + NetError::Io(_) | NetError::Disconnected => true, + NetError::Encode(s) => { + let l = s.to_ascii_lowercase(); + l.contains("broken pipe") + || l.contains("connection reset") + || l.contains("connection closed") + || l.contains("i2p sam connect") + || l.contains("early eof") + || l.contains("unexpected eof") + } + _ => false, + } +} + +fn stream_retry(err: &NetError) -> bool { + match err { + NetError::Io(_) | NetError::Disconnected => true, + NetError::Encode(s) => { + let l = s.to_ascii_lowercase(); + l.contains("early eof") + || l.contains("unexpected eof") + || l.contains("broken pipe") + || l.contains("connection reset") + || l.contains("connection closed") + || l.contains("cant_reach") + || l.contains("timeout") + } + _ => false, + } +} + #[cfg(test)] fn clear_installed() { *INSTALLED.lock().unwrap_or_else(|e| e.into_inner()) = None; @@ -125,6 +164,24 @@ impl I2pSam { async fn connect_session_dest( sam_addr: SocketAddr, dest: Option<&str>, + ) -> Result<(Self, String), NetError> { + let mut last = None; + for _ in 0..32 { + match Self::connect_session_dest_once(sam_addr, dest).await { + Ok(v) => return Ok(v), + Err(e) if sam_retry(&e) => { + last = Some(e); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + Err(e) => return Err(e), + } + } + Err(last.expect("sam retry")) + } + + async fn connect_session_dest_once( + sam_addr: SocketAddr, + dest: Option<&str>, ) -> Result<(Self, String), NetError> { let mut control = TcpStream::connect(sam_addr) .await @@ -132,11 +189,7 @@ impl I2pSam { hello(&mut control).await?; let session_id = fresh_session_id(); let dest_arg = dest.unwrap_or("TRANSIENT"); - write_line( - &mut control, - &format!("SESSION CREATE STYLE=STREAM ID={session_id} DESTINATION={dest_arg}"), - ) - .await?; + write_line(&mut control, &session_create_line(&session_id, dest_arg)).await?; let reply = read_line(&mut control).await?; if !reply.to_ascii_uppercase().contains("RESULT=OK") { return Err(NetError::Encode(format!("i2p sam session: {reply}"))); @@ -151,6 +204,7 @@ impl I2pSam { sam_addr, session_id, destination: destination.clone(), + forward_port: None, _control: control, _forward: None, }, @@ -167,35 +221,77 @@ impl I2pSam { sam_addr: self.sam_addr, session_id: self.session_id.clone(), destination: self.destination.clone(), + forward_port: self.forward_port, } } + /// BIP155 address of this session. Port is 0: SAM 3.1 does not use ports. + pub fn local_netaddr(&self) -> Result { + i2p_addr_from_sam_b64(&self.destination) + } + fn into_keepalive(self) -> (I2pDialer, TcpStream) { let dialer = self.dialer(); (dialer, self._control) } pub async fn stream_forward(&mut self, port: u16) -> Result<(), NetError> { + let mut last = None; + for _ in 0..24 { + match self.stream_forward_once(port).await { + Ok(s) => { + self.forward_port = Some(port); + self._forward = Some(s); + return Ok(()); + } + Err(e) if stream_retry(&e) => { + last = Some(e); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + Err(e) => return Err(e), + } + } + Err(last.expect("forward retry")) + } + + async fn stream_forward_once(&self, port: u16) -> Result { let mut s = TcpStream::connect(self.sam_addr) .await .map_err(|e| NetError::Encode(format!("i2p sam forward connect: {e}")))?; hello(&mut s).await?; write_line( &mut s, - &format!("STREAM FORWARD ID={} PORT={port}", self.session_id), + &format!( + "STREAM FORWARD ID={} PORT={port} HOST=127.0.0.1 SILENT=true", + self.session_id + ), ) .await?; let reply = read_line(&mut s).await?; if !reply.to_ascii_uppercase().contains("RESULT=OK") { return Err(NetError::Encode(format!("i2p sam forward: {reply}"))); } - self._forward = Some(s); - Ok(()) + Ok(s) } } impl I2pDialer { pub async fn stream_connect(&self, dest_b32: &str) -> Result { + let mut last = None; + for _ in 0..24 { + match self.stream_connect_once(dest_b32).await { + Ok(s) => return Ok(s), + Err(e) if stream_retry(&e) => { + last = Some(e); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + Err(e) => return Err(e), + } + } + Err(last.expect("stream retry")) + } + + async fn stream_connect_once(&self, dest_b32: &str) -> Result { let mut s = TcpStream::connect(self.sam_addr) .await .map_err(|e| NetError::Encode(format!("i2p sam stream connect: {e}")))?; @@ -215,12 +311,97 @@ impl I2pDialer { Ok(s) } - async fn recreate_session(&self) -> Result<(Self, TcpStream), NetError> { - let (sam, _) = I2pSam::connect_session_dest(self.sam_addr, Some(&self.destination)).await?; - Ok(sam.into_keepalive()) + async fn recreate_session(&self) -> Result<(Self, TcpStream, Option), NetError> { + let (mut sam, _) = + I2pSam::connect_session_dest(self.sam_addr, Some(&self.destination)).await?; + let forward = match self.forward_port { + Some(port) => { + sam.stream_forward(port).await?; + sam._forward.take() + } + None => None, + }; + let (dialer, keepalive) = sam.into_keepalive(); + Ok((dialer, keepalive, forward)) } } +// Ed25519 (7), not the SAM DSA_SHA1 default. leaseSetEncType 4 is ECIES. +fn session_create_line(session_id: &str, dest_arg: &str) -> String { + let sig = if dest_arg.eq_ignore_ascii_case("TRANSIENT") { + " SIGNATURE_TYPE=7" + } else { + "" + }; + format!( + "SESSION CREATE STYLE=STREAM ID={session_id} DESTINATION={dest_arg}{sig} \ + i2cp.leaseSetEncType=4,0 inbound.length=1 outbound.length=1 \ + inbound.quantity=1 outbound.quantity=1" + ) +} + +// Public Destination is 387 bytes plus the cert length at bytes 385–386. +// The trailing private key is not part of the BIP155 hash. +fn i2p_addr_from_sam_b64(dest_b64: &str) -> Result { + let raw = decode_i2p_b64(dest_b64.trim())?; + let public = i2p_public_destination(&raw)?; + let digest = sha256::Hash::hash(public); + Ok(crate::NetAddr::I2p { + dest: *digest.as_byte_array(), + port: 0, + }) +} + +fn i2p_public_destination(raw: &[u8]) -> Result<&[u8], NetError> { + const DEST_LEN_BASE: usize = 387; + const CERT_LEN_POS: usize = 385; + if raw.len() < CERT_LEN_POS + 2 { + return Err(NetError::Encode(format!( + "i2p destination too short ({})", + raw.len() + ))); + } + let cert_len = u16::from_be_bytes([raw[CERT_LEN_POS], raw[CERT_LEN_POS + 1]]) as usize; + let dest_len = DEST_LEN_BASE + cert_len; + if dest_len > raw.len() { + return Err(NetError::Encode(format!( + "i2p certificate length {cert_len} needs {dest_len} bytes, have {}", + raw.len() + ))); + } + Ok(&raw[..dest_len]) +} + +fn decode_i2p_b64(s: &str) -> Result, NetError> { + let mut bytes = Vec::with_capacity(s.len() * 3 / 4); + let mut acc = 0u32; + let mut n = 0u32; + for c in s.chars() { + if c == '=' { + break; + } + let v = match c { + 'A'..='Z' => c as u32 - 'A' as u32, + 'a'..='z' => c as u32 - 'a' as u32 + 26, + '0'..='9' => c as u32 - '0' as u32 + 52, + '-' | '+' => 62, + '~' | '/' => 63, + _ => { + return Err(NetError::Encode(format!( + "i2p destination base64 contains {c:?}" + ))); + } + }; + acc = (acc << 6) | v; + n += 6; + if n >= 8 { + n -= 8; + bytes.push((acc >> n) as u8); + } + } + Ok(bytes) +} + fn sam_kv<'a>(line: &'a str, key: &str) -> Option<&'a str> { let prefix = format!("{key}="); line.split_whitespace().find_map(|tok| { @@ -285,16 +466,23 @@ async fn write_line(s: &mut TcpStream, line: &str) -> Result<(), NetError> { } async fn read_line(s: &mut TcpStream) -> Result { - let mut reader = BufReader::new(s); - let mut line = String::new(); - let n = reader - .read_line(&mut line) - .await - .map_err(|e| NetError::Encode(format!("i2p sam read: {e}")))?; - if n == 0 { - return Err(NetError::Encode("i2p sam: connection closed".into())); + let mut buf = Vec::new(); + loop { + let mut b = [0u8; 1]; + s.read_exact(&mut b) + .await + .map_err(|e| NetError::Encode(format!("i2p sam read: {e}")))?; + if b[0] == b'\n' { + break; + } + if b[0] != b'\r' { + buf.push(b[0]); + } + if buf.len() > 16 * 1024 { + return Err(NetError::Encode("i2p sam read: line too long".into())); + } } - Ok(line.trim_end_matches(['\r', '\n']).to_string()) + String::from_utf8(buf).map_err(|e| NetError::Encode(format!("i2p sam read: {e}"))) } #[cfg(test)] @@ -311,6 +499,21 @@ mod tests { async fn fake_sam( ok_hello: bool, dest_log: Arc>>, + ) -> (SocketAddr, Arc>>) { + fake_sam_opts( + ok_hello, + dest_log, + Arc::new(Mutex::new(0)), + Arc::new(Mutex::new(0)), + ) + .await + } + + async fn fake_sam_opts( + ok_hello: bool, + dest_log: Arc>>, + stream_drops: Arc>, + forward_drops: Arc>, ) -> (SocketAddr, Arc>>) { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -323,6 +526,8 @@ mod tests { }; let log = Arc::clone(&dest_log); let live = Arc::clone(&live_accept); + let drops = Arc::clone(&stream_drops); + let fwd_drops = Arc::clone(&forward_drops); let ok = ok_hello; tokio::spawn(async move { let mut created_id: Option = None; @@ -359,9 +564,33 @@ mod tests { .await; } else if up.starts_with("STREAM FORWARD") { log.lock().unwrap().push(line.clone()); + let drop_n = { + let mut n = fwd_drops.lock().unwrap(); + if *n > 0 { + *n -= 1; + true + } else { + false + } + }; + if drop_n { + break; + } let _ = write_line(&mut s, "STREAM STATUS RESULT=OK").await; } else if up.starts_with("STREAM CONNECT") { log.lock().unwrap().push(line.clone()); + let drop_n = { + let mut n = drops.lock().unwrap(); + if *n > 0 { + *n -= 1; + true + } else { + false + } + }; + if drop_n { + break; + } let id = sam_kv(&line, "ID").unwrap_or(""); let known = live.lock().unwrap().contains(id); if known { @@ -392,6 +621,31 @@ mod tests { .collect() } + #[tokio::test] + async fn i2p_sam_stream_connect_retries_early_eof() { + let log = Arc::new(Mutex::new(Vec::new())); + let drops = Arc::new(Mutex::new(2)); + let (addr, _live) = + fake_sam_opts(true, Arc::clone(&log), drops, Arc::new(Mutex::new(0))).await; + let sam = I2pSam::connect(addr).await.unwrap(); + let dest = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa.b32.i2p"; + sam.stream_connect(dest).await.unwrap(); + let got = stream_lines(&log, "STREAM CONNECT"); + assert_eq!(got.len(), 3, "{got:?}"); + } + + #[tokio::test] + async fn i2p_sam_stream_forward_retries_early_eof() { + let log = Arc::new(Mutex::new(Vec::new())); + let drops = Arc::new(Mutex::new(2)); + let (addr, _live) = + fake_sam_opts(true, Arc::clone(&log), Arc::new(Mutex::new(0)), drops).await; + let mut sam = I2pSam::connect(addr).await.unwrap(); + sam.stream_forward(18444).await.unwrap(); + let got = stream_lines(&log, "STREAM FORWARD"); + assert_eq!(got.len(), 3, "{got:?}"); + } + #[tokio::test] async fn i2p_sam_stream_connect_fake() { let log = Arc::new(Mutex::new(Vec::new())); @@ -405,6 +659,10 @@ mod tests { for g in &got { assert!(g.contains(dest), "{g}"); assert!(g.contains("ID=rbtc"), "{g}"); + // i2pd drops the STREAM STATUS line when CONNECT sets SILENT=true, + // so the dial never sees RESULT=OK. FORWARD is the command that + // prefixes the peer destination; CONNECT must stay non-silent. + assert!(!g.to_ascii_uppercase().contains("SILENT=TRUE"), "{g}"); } let (bad, _live) = fake_sam(false, Arc::new(Mutex::new(Vec::new()))).await; @@ -468,9 +726,16 @@ mod tests { assert_eq!(fw.len(), 1, "{fw:?}"); assert!(fw[0].contains("PORT=18444"), "{}", fw[0]); assert!(fw[0].contains("ID=rbtc"), "{}", fw[0]); + assert!(fw[0].contains("HOST=127.0.0.1"), "{}", fw[0]); + assert!(fw[0].contains("SILENT=true"), "{}", fw[0]); let creates = stream_lines(&log, "SESSION CREATE"); assert!( - creates.iter().any(|c| c.contains("DESTINATION=TRANSIENT")), + creates.iter().any(|c| { + c.contains("DESTINATION=TRANSIENT") + && c.contains("SIGNATURE_TYPE=7") + && c.contains("i2cp.leaseSetEncType=4,0") + && c.contains("inbound.length=1") + }), "{creates:?}" ); @@ -478,9 +743,11 @@ mod tests { sam2.stream_forward(18444).await.unwrap(); let creates = stream_lines(&log, "SESSION CREATE"); assert!( - creates - .iter() - .any(|c| c.contains(&format!("DESTINATION={FAKE_DEST}"))), + creates.iter().any(|c| { + c.contains(&format!("DESTINATION={FAKE_DEST}")) + && c.contains("i2cp.leaseSetEncType=4,0") + && !c.contains("SIGNATURE_TYPE=") + }), "{creates:?}" ); let _ = std::fs::remove_dir_all(&dir); @@ -528,4 +795,105 @@ mod tests { ); clear_installed(); } + + #[tokio::test] + async fn i2p_sam_recreate_restores_forward() { + let _gate = INSTALL_GATE.lock().await; + clear_installed(); + let log = Arc::new(Mutex::new(Vec::new())); + let (addr, live) = fake_sam(true, Arc::clone(&log)).await; + let mut sam = I2pSam::connect(addr).await.unwrap(); + sam.stream_forward(18444).await.unwrap(); + crate::socks::install_i2p_dialer(sam.dialer()); + let peer = crate::NetAddr::I2p { + dest: [0u8; 32], + port: 0, + }; + crate::socks::Dialer::Direct + .connect_net(peer) + .await + .unwrap(); + assert_eq!(stream_lines(&log, "STREAM FORWARD").len(), 1); + live.lock().unwrap().clear(); + crate::socks::Dialer::Direct + .connect_net(peer) + .await + .expect("INVALID_ID must recreate the session"); + let fw = stream_lines(&log, "STREAM FORWARD"); + assert_eq!(fw.len(), 2, "dropped session must FORWARD again: {fw:?}"); + assert!(fw[1].contains("PORT=18444"), "{}", fw[1]); + assert!(fw[1].contains("SILENT=true"), "{}", fw[1]); + clear_installed(); + } + + fn i2p_b64_encode(raw: &[u8]) -> String { + const ALPH: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-~"; + let mut out = String::new(); + let mut i = 0; + while i < raw.len() { + let b0 = raw[i]; + let b1 = if i + 1 < raw.len() { raw[i + 1] } else { 0 }; + let b2 = if i + 2 < raw.len() { raw[i + 2] } else { 0 }; + out.push(ALPH[(b0 >> 2) as usize] as char); + out.push(ALPH[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char); + if i + 1 < raw.len() { + out.push(ALPH[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char); + } + if i + 2 < raw.len() { + out.push(ALPH[(b2 & 63) as usize] as char); + } + i += 3; + } + out + } + + #[test] + fn i2p_addr_hashes_public_destination_only() { + let mut ident = vec![7u8; 387]; + ident[384] = 0; + ident[385] = 0; + ident[386] = 0; + let mut privd = ident.clone(); + privd.extend_from_slice(&[9u8; 80]); + let a = i2p_addr_from_sam_b64(&i2p_b64_encode(&ident)).unwrap(); + let b = i2p_addr_from_sam_b64(&i2p_b64_encode(&privd)).unwrap(); + let want = crate::NetAddr::I2p { + dest: [ + 0x15, 0xe2, 0xe8, 0x24, 0x7c, 0x9d, 0xfa, 0x2e, 0xd0, 0xaa, 0x88, 0x04, 0xf8, 0x36, + 0xfc, 0x3a, 0x53, 0x23, 0x0d, 0x3c, 0xa5, 0x8e, 0x03, 0x4d, 0x05, 0x7f, 0xb0, 0xee, + 0x90, 0x7b, 0x3a, 0x57, + ], + port: 0, + }; + assert_eq!(a, want); + assert_eq!(b, want, "private key suffix must not change the b32"); + assert_eq!( + a.to_string(), + "cxroqjd4tx5c5ufkracpqnx4hjjsgdj4uwhagtifp6yo5ed3hjlq.b32.i2p:0" + ); + + let mut public = vec![9u8; 387]; + public[384] = 5; + public[385] = 0; + public[386] = 4; + public.extend_from_slice(&[0x00, 0x07, 0x00, 0x04]); + let mut full = public.clone(); + full.extend_from_slice(&[0xab; 64]); + let ed = i2p_addr_from_sam_b64(&i2p_b64_encode(&public)).unwrap(); + let ed_priv = i2p_addr_from_sam_b64(&i2p_b64_encode(&full)).unwrap(); + assert_eq!(ed, ed_priv); + assert_eq!( + ed, + crate::NetAddr::I2p { + dest: [ + 0x0f, 0x35, 0x9f, 0x01, 0x19, 0x78, 0x92, 0x09, 0x23, 0x60, 0xcb, 0x0d, 0x3e, + 0xeb, 0x27, 0xcf, 0x5f, 0xf7, 0xb7, 0x1f, 0x70, 0xef, 0xe7, 0xf7, 0x8d, 0x45, + 0x03, 0x1e, 0x1a, 0xe2, 0x11, 0x75, + ], + port: 0, + } + ); + assert!(i2p_addr_from_sam_b64("!!!").is_err()); + assert!(i2p_addr_from_sam_b64(&i2p_b64_encode(&[1, 2, 3])).is_err()); + } } diff --git a/crates/rbitcoin-net/src/ibd/dial.rs b/crates/rbitcoin-net/src/ibd/dial.rs index 31437abbb..2ce328873 100644 --- a/crates/rbitcoin-net/src/ibd/dial.rs +++ b/crates/rbitcoin-net/src/ibd/dial.rs @@ -280,9 +280,10 @@ pub(crate) async fn dial_batch( trying_connection_log(PeerConnType::OutboundFullRelay, addr) ); let dialer = dialer.clone(); + let to = connect_timeout_for(addr, connect_timeout); handles.push(tokio::spawn(async move { let fut = spawn_peer(id, addr, magic, local_addr, tip_h, sinks, dialer); - match tokio::time::timeout(connect_timeout, fut).await { + match tokio::time::timeout(to, fut).await { Ok(Ok(slot)) => Ok(slot), Ok(Err(e)) => { let kind = classify_dial_err(&e); @@ -292,7 +293,7 @@ pub(crate) async fn dial_batch( id, addr, DialFailKind::Network, - format!("connect timeout ({connect_timeout:?})"), + format!("connect timeout ({to:?})"), )), } })); @@ -317,6 +318,14 @@ pub(crate) async fn dial_batch( out } +/// I2P STREAM CONNECT waits on tunnel + leaseset lookup; 8s is a clearnet RTT. +pub fn connect_timeout_for(addr: crate::NetAddr, base: Duration) -> Duration { + match addr { + crate::NetAddr::I2p { .. } => base.max(Duration::from_secs(90)), + _ => base, + } +} + /// How many new dials to start when below `target` live peers. /// /// At least 2 so a single lemon cannot occupy the only spare slot forever; @@ -640,6 +649,18 @@ mod tests { SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, o)), 8333) } + #[test] + fn i2p_connect_timeout_is_longer_than_clearnet() { + let i2p = crate::NetAddr::I2p { + dest: [0u8; 32], + port: 1, + }; + let ip: crate::NetAddr = "127.0.0.1:1".parse().unwrap(); + let base = Duration::from_secs(8); + assert_eq!(connect_timeout_for(ip, base), base); + assert_eq!(connect_timeout_for(i2p, base), Duration::from_secs(90)); + } + fn dummy_slot(id: usize, a: SocketAddr, alive: bool) -> PeerSlot { let (cmd_tx, _rx) = tokio::sync::mpsc::unbounded_channel(); // JoinHandle without a running runtime: abort on Drop is still safe. diff --git a/crates/rbitcoin-net/src/ibd/mod.rs b/crates/rbitcoin-net/src/ibd/mod.rs index 6b209f227..de6afe45b 100644 --- a/crates/rbitcoin-net/src/ibd/mod.rs +++ b/crates/rbitcoin-net/src/ibd/mod.rs @@ -31,6 +31,7 @@ mod reorg; mod state; mod status; +pub use dial::connect_timeout_for; pub use perf_log::{format_tip_perf_sizes, read_proc_rss, ProcRss, TipPerfSizes}; use archive::{rehydrate_block_queue_into_confirm, rehydrate_class_a_into_body_queue}; diff --git a/crates/rbitcoin-net/src/lib.rs b/crates/rbitcoin-net/src/lib.rs index 35ac3a9ec..583229b7a 100644 --- a/crates/rbitcoin-net/src/lib.rs +++ b/crates/rbitcoin-net/src/lib.rs @@ -40,8 +40,8 @@ pub use error::NetError; pub use eviction::{select_inbound_eviction, InboundEvictCandidate}; pub use i2p_sam::I2pSam; pub use ibd::{ - format_tip_perf_sizes, read_proc_rss, rehydrate_block_queue_residue, IbdConfig, ProcRss, - TipPerfSizes, DEFAULT_BLOCKS_IN_TRANSIT_PER_PEER, DEFAULT_IBD_WINDOW, + connect_timeout_for, format_tip_perf_sizes, read_proc_rss, rehydrate_block_queue_residue, + IbdConfig, ProcRss, TipPerfSizes, DEFAULT_BLOCKS_IN_TRANSIT_PER_PEER, DEFAULT_IBD_WINDOW, }; pub use most_work::sum_work; pub use net_permissions::{ diff --git a/crates/rbitcoin-net/src/netaddr.rs b/crates/rbitcoin-net/src/netaddr.rs index 408728951..a4a550802 100644 --- a/crates/rbitcoin-net/src/netaddr.rs +++ b/crates/rbitcoin-net/src/netaddr.rs @@ -88,19 +88,18 @@ impl FromStr for NetAddr { impl NetAddr { pub fn from_addrv2(msg: &AddrV2Message) -> Option { - if msg.port == 0 { - return None; - } match &msg.addr { + // SAM 3.1 has no ports. Core advertises I2P with port 0. + AddrV2::I2p(dest) => Some(NetAddr::I2p { + dest: *dest, + port: msg.port, + }), + _ if msg.port == 0 => None, AddrV2::Ipv4(_) | AddrV2::Ipv6(_) => msg.socket_addr().ok().map(NetAddr::from_socket), AddrV2::TorV3(pk) => Some(NetAddr::Onion { pk: *pk, port: msg.port, }), - AddrV2::I2p(dest) => Some(NetAddr::I2p { - dest: *dest, - port: msg.port, - }), AddrV2::Cjdns(ip) if is_cjdns_ip(*ip) => Some(NetAddr::Cjdns { ip: *ip, port: msg.port, @@ -433,6 +432,31 @@ mod tests { assert!("short.b32.i2p:1".parse::().is_err()); } + /// Core SAM 3.1 advertises I2P with port 0. Other networks still reject port 0. + #[test] + fn netaddr_i2p_port_zero_is_core_sam() { + let dest = [0x11u8; 32]; + let msg = AddrV2Message { + time: 1, + services: bitcoin::p2p::ServiceFlags::NETWORK, + addr: AddrV2::I2p(dest), + port: 0, + }; + let a = NetAddr::from_addrv2(&msg).expect("core i2p port 0"); + assert_eq!(a, NetAddr::I2p { dest, port: 0 }); + assert_eq!(a.port(), 0); + let s = a.to_string(); + assert!(s.ends_with(".b32.i2p:0"), "{s}"); + assert_eq!(s.parse::().unwrap(), a); + let onion = AddrV2Message { + time: 1, + services: bitcoin::p2p::ServiceFlags::NETWORK, + addr: AddrV2::TorV3([0x22; 32]), + port: 0, + }; + assert!(NetAddr::from_addrv2(&onion).is_none()); + } + #[test] fn netaddr_cjdns_addrv2_roundtrip() { let ip = Ipv6Addr::new(0xfc00, 1, 2, 3, 4, 5, 6, 7); diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index 0cea54c68..6501c1e33 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -5660,6 +5660,58 @@ fn self_announce_onion_not_external_ip() { ); } +#[test] +fn self_announce_i2p_not_external_ip() { + use crate::peers::{PeerConnType, PeerHub}; + use bitcoin::p2p::address::{AddrV2, Address}; + use bitcoin::p2p::message::NetworkMessage; + use bitcoin::p2p::message_network::VersionMessage; + use bitcoin::p2p::ServiceFlags; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + + let hub = PeerHub::new(); + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18444); + let ver = VersionMessage { + version: 70016, + services: ServiceFlags::NETWORK, + timestamp: 0, + receiver: Address::new(&addr, ServiceFlags::NONE), + sender: Address::new(&addr, ServiceFlags::NONE), + nonce: 1, + user_agent: "/rbitcoin:test/".into(), + start_height: 0, + relay: true, + }; + let peer = hub.register(addr, addr, &ver, false, PeerConnType::OutboundFullRelay); + peer.set_wants_addrv2(); + let i2p = crate::NetAddr::I2p { + dest: [0x11; 32], + port: 0, + }; + hub.set_p2p_i2p(i2p); + hub.set_discover(false); + hub.set_clearnet_listen(false); + hub.set_external_ips(vec![IpAddr::V4(Ipv4Addr::new(42, 42, 42, 42))]); + hub.set_listen_port(18445); + match peer.take_self_announce_msg().expect("i2p announce") { + NetworkMessage::AddrV2(v) => { + assert_eq!(v.len(), 1, "{v:?}"); + assert!(matches!(v[0].addr, AddrV2::I2p(d) if d == [0x11; 32])); + assert_eq!(v[0].port, 0); + } + other => panic!("expected AddrV2 i2p, got {other:?}"), + } + let rows = hub.rpc_local_addresses(); + assert_eq!(rows.len(), 1, "{rows:?}"); + assert_eq!(rows[0].0, i2p.host_str()); + assert_eq!(rows[0].1, 0); + let v1 = hub.register(addr, addr, &ver, false, PeerConnType::OutboundFullRelay); + assert!( + v1.take_self_announce_msg().is_none(), + "v1 ADDR cannot carry an I2P destination" + ); +} + #[test] fn redundant_verack_is_ignored_and_logged() { use bitcoin::consensus::encode::serialize; diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index e9bdcccba..11564850b 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -446,8 +446,9 @@ impl LivePeer { } let hub = self.owner.upgrade()?; let onion = hub.p2p_onion(); + let i2p = hub.p2p_i2p(); let sock = hub.advertise_local_socket(); - if onion.is_none() && sock.is_none() { + if onion.is_none() && i2p.is_none() && sock.is_none() { return None; } const DAY: u64 = 24 * 60 * 60; @@ -473,6 +474,15 @@ impl LivePeer { port: addr.port(), }); } + if let Some(addr) = i2p { + rbitcoin_log::debug!("{}", crate::peer::advertising_address_log(addr, self.id)); + v.push(AddrV2Message { + time: t, + services, + addr: addr.to_addrv2(), + port: addr.port(), + }); + } if let Some(sock) = sock { rbitcoin_log::debug!("{}", crate::peer::advertising_address_log(sock, self.id)); let net = crate::NetAddr::from_socket(sock); @@ -1142,6 +1152,7 @@ pub struct PeerHub { external_ips: Mutex>, wallet_onions: Mutex>, p2p_onion: Mutex>, + p2p_i2p: Mutex>, /// P2P listen port used with advertised external IPs. listen_port: AtomicU16, /// Core `-discover`. Off: never self-announce, even with `--external-ip`. @@ -1232,6 +1243,7 @@ impl PeerHub { external_ips: Mutex::new(Vec::new()), wallet_onions: Mutex::new(Vec::new()), p2p_onion: Mutex::new(None), + p2p_i2p: Mutex::new(None), listen_port: AtomicU16::new(0), discover: AtomicBool::new(true), clearnet_listen: AtomicBool::new(true), @@ -1311,6 +1323,16 @@ impl PeerHub { format!("{host}:{port}").parse().ok() } + pub fn set_p2p_i2p(&self, addr: crate::NetAddr) { + if matches!(addr, crate::NetAddr::I2p { .. }) { + *self.p2p_i2p.lock().unwrap_or_else(|e| e.into_inner()) = Some(addr); + } + } + + pub fn p2p_i2p(&self) -> Option { + *self.p2p_i2p.lock().unwrap_or_else(|e| e.into_inner()) + } + pub fn set_clearnet_listen(&self, on: bool) { self.clearnet_listen.store(on, Ordering::Relaxed); } @@ -1346,6 +1368,9 @@ impl PeerHub { { rows.push((host, port, LOCAL_MANUAL)); } + if let Some(addr) = self.p2p_i2p() { + rows.push((addr.host_str(), addr.port(), LOCAL_MANUAL)); + } if !self.discover.load(Ordering::Relaxed) { return rows; } @@ -1753,7 +1778,13 @@ impl PeerHub { inbound: bool, conn_type: PeerConnType, ) -> Arc { - self.register_connecting_net(addr, crate::NetAddr::Ip(addr), addrbind, inbound, conn_type) + self.register_connecting_net( + addr, + crate::NetAddr::from_socket(addr), + addrbind, + inbound, + conn_type, + ) } pub fn register_connecting_net( @@ -1801,7 +1832,7 @@ impl PeerHub { ) -> Arc { self.register_net( addr, - crate::NetAddr::Ip(addr), + crate::NetAddr::from_socket(addr), addrbind, ver, inbound, @@ -1849,7 +1880,7 @@ impl PeerHub { id, PeerEndpoint { addr, - net: crate::NetAddr::Ip(addr), + net: crate::NetAddr::from_socket(addr), addrbind, }, ver, @@ -3327,6 +3358,29 @@ mod tests { ); } + #[test] + fn learn_addrv2_keeps_i2p_port_zero() { + use bitcoin::p2p::address::{AddrV2, AddrV2Message}; + + let hub = PeerHub::new(); + let am = Arc::new(Mutex::new(crate::seeds::AddrMan::new())); + hub.set_addrman(am.clone()); + let dest = [0x22u8; 32]; + hub.learn_addrv2(&[AddrV2Message { + time: 1, + services: ServiceFlags::NETWORK, + addr: AddrV2::I2p(dest), + port: 0, + }]); + let book = am.lock().unwrap_or_else(|e| e.into_inner()); + let want = crate::NetAddr::I2p { dest, port: 0 }; + assert!( + book.entries().iter().any(|e| e.addr == want), + "Core I2P port 0 must stay in the book, got {:?}", + book.entries() + ); + } + #[test] fn dial_target_cjdns_is_native_socket() { use std::net::Ipv6Addr; @@ -3382,6 +3436,36 @@ mod tests { ); } + #[test] + fn inbound_fc00_getpeerinfo_network_is_cjdns() { + use bitcoin::p2p::address::Address; + use bitcoin::p2p::message_network::VersionMessage; + use bitcoin::p2p::ServiceFlags; + use std::net::Ipv6Addr; + + let ip = Ipv6Addr::new(0xfc00, 1, 2, 3, 4, 5, 6, 7); + let addr = SocketAddr::from((ip, 8333)); + let bind = SocketAddr::from((ip, 18444)); + let hub = PeerHub::new(); + let connecting = hub.register_connecting(addr, bind, true, PeerConnType::Inbound); + assert_eq!(connecting.net.network_label(), "cjdns"); + hub.unregister(connecting.id); + let ver = VersionMessage { + version: bitcoin::p2p::PROTOCOL_VERSION, + services: ServiceFlags::NETWORK, + timestamp: 0, + receiver: Address::new(&addr, ServiceFlags::NONE), + sender: Address::new(&bind, ServiceFlags::NONE), + nonce: 0, + user_agent: "/rbitcoin:test/".into(), + start_height: 3, + relay: true, + }; + let live = hub.register_with_id(1, addr, bind, &ver, true, PeerConnType::Inbound); + assert_eq!(live.net.network_label(), "cjdns"); + assert_eq!(hub.snapshot()[0].net.network_label(), "cjdns"); + } + #[test] fn fc00_not_advertisable_without_cjdns_reachable() { use std::net::Ipv6Addr; diff --git a/crates/rbitcoin-node/src/lib.rs b/crates/rbitcoin-node/src/lib.rs index dd24d8009..fafb39c74 100644 --- a/crates/rbitcoin-node/src/lib.rs +++ b/crates/rbitcoin-node/src/lib.rs @@ -10,6 +10,8 @@ mod run; mod tor_control; pub use cli::cli_main; -pub use config::{DatadirOpts, ListenOpts, MempoolOpts, NodeConfig, RpcOpts, TorControlOpts}; +pub use config::{ + DatadirOpts, ListenOpts, MempoolOpts, NodeConfig, P2pListen, RpcOpts, TorControlOpts, +}; pub use error::NodeError; pub use run::{run_node, run_p2p, NodeHandle}; diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index ea44e529c..b6ba607f1 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -100,6 +100,14 @@ impl Shutdown { } } +/// Clearnet follow handshake bound. Overlay STREAM CONNECT is slower; see +/// [`follow_connect_timeout`]. +const FOLLOW_CONNECT_SECS: u64 = 8; + +fn follow_connect_timeout(peer: rbitcoin_net::NetAddr) -> Duration { + rbitcoin_net::connect_timeout_for(peer, Duration::from_secs(FOLLOW_CONNECT_SECS)) +} + /// Install SIGTERM / SIGINT (and Ctrl+C) handlers that trip `shutdown`. fn spawn_signal_handler(shutdown: Arc) { tokio::spawn(async move { @@ -325,9 +333,13 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { if node.local_addr.port() != 0 { node.peers.set_listen_port(node.local_addr.port()); } - if !config.listen.external_ips.is_empty() { - node.peers - .set_external_ips(config.listen.external_ips.clone()); + let external = external_ips_with_cjdns_bind( + &config.listen.external_ips, + node.local_addr, + config.listen.cjdns_reachable, + ); + if !external.is_empty() { + node.peers.set_external_ips(external); } if immediate_relay { node.peers.set_noban(true); @@ -408,6 +420,11 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { sam.stream_forward(port) .await .map_err(|e| NodeError::Init(format!("i2p STREAM FORWARD {port}: {e}")))?; + let local = sam + .local_netaddr() + .map_err(|e| NodeError::Init(format!("i2p address: {e}")))?; + info!("i2p address {local}"); + node.peers.set_p2p_i2p(local); info!("i2p STREAM FORWARD to {}", node.local_addr); } if let Some(sam) = i2p_sam.as_ref() { @@ -625,7 +642,6 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { if tip_follow_ready && !shutdown.requested() { let follow_n = targets.len().min(max_out.min(3)); - const FOLLOW_CONNECT_SECS: u64 = 8; if catch_up.dial_failed_all() { for peer in targets.iter().take(follow_n) { if let Err(e) = node.peers.dial_net(*peer, PeerConnType::OutboundFullRelay) { @@ -640,16 +656,14 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { if shutdown.requested() { break; } + let to = follow_connect_timeout(*peer); tokio::select! { biased; _ = shutdown.cancelled() => { warn!("signal: skip remaining follow connects"); break; } - result = tokio::time::timeout( - Duration::from_secs(FOLLOW_CONNECT_SECS), - node.follow_from_net(*peer), - ) => { + result = tokio::time::timeout(to, node.follow_from_net(*peer)) => { match result { Ok(Ok(())) => { info!( @@ -658,9 +672,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { ); } Ok(Err(e)) => warn!("node: follow {peer} failed: {e}"), - Err(_) => warn!( - "node: follow {peer} timed out ({FOLLOW_CONNECT_SECS}s)" - ), + Err(_) => warn!("node: follow {peer} timed out ({to:?})"), } } } @@ -1863,6 +1875,27 @@ pub(crate) async fn tip_follow_next_wake( } } +/// A CJDNS listen bind is the address peers should learn. `--external-ip` +/// still wins when the operator set the same address already. +fn external_ips_with_cjdns_bind( + configured: &[std::net::IpAddr], + bind: SocketAddr, + cjdns_reachable: bool, +) -> Vec { + let mut ips = configured.to_vec(); + if !cjdns_reachable { + return ips; + } + let std::net::IpAddr::V6(ip) = bind.ip() else { + return ips; + }; + if !rbitcoin_net::is_cjdns_ip(ip) || ips.contains(&std::net::IpAddr::V6(ip)) { + return ips; + } + ips.push(std::net::IpAddr::V6(ip)); + ips +} + /// Parse Core `-seednode` host or host:port using the chain default P2P port. fn resolve_seednode(raw: &str, network: Network) -> Result { if let Ok(a) = raw.parse::() { @@ -2121,6 +2154,42 @@ mod tests { assert_eq!(catch_up_after_err(10, true, true), CatchUp::Incomplete); } + #[test] + fn cjdns_listen_is_advertised_without_external_ip() { + use std::net::{IpAddr, Ipv6Addr}; + let fc = Ipv6Addr::new(0xfc00, 1, 2, 3, 4, 5, 6, 7); + let bind = SocketAddr::from((fc, 8333)); + let got = external_ips_with_cjdns_bind(&[], bind, true); + assert_eq!(got, vec![IpAddr::V6(fc)]); + assert!( + external_ips_with_cjdns_bind(&[], bind, false).is_empty(), + "fc00 stays unroutable until --cjdns-reachable" + ); + let v4 = SocketAddr::from(([1, 2, 3, 4], 8333)); + assert!(external_ips_with_cjdns_bind(&[], v4, true).is_empty()); + let ula = SocketAddr::from((Ipv6Addr::new(0xfd00, 0, 0, 0, 0, 0, 0, 1), 8333)); + assert!( + external_ips_with_cjdns_bind(&[], ula, true).is_empty(), + "fd00::/8 is not CJDNS" + ); + let dup = external_ips_with_cjdns_bind(&[IpAddr::V6(fc)], bind, true); + assert_eq!(dup, vec![IpAddr::V6(fc)]); + } + + #[test] + fn follow_connect_timeout_i2p_is_longer_than_clearnet() { + let i2p = rbitcoin_net::NetAddr::I2p { + dest: [0u8; 32], + port: 1, + }; + let ip: rbitcoin_net::NetAddr = "127.0.0.1:1".parse().unwrap(); + assert_eq!( + follow_connect_timeout(ip), + Duration::from_secs(FOLLOW_CONNECT_SECS) + ); + assert_eq!(follow_connect_timeout(i2p), Duration::from_secs(90)); + } + #[test] fn catch_up_retry_config_uses_production_not_for_test() { let peers = std::sync::Arc::new(std::sync::Mutex::new(rbitcoin_net::AddrMan::new())); diff --git a/crates/rbitcoin-rpc/src/methods_tests.rs b/crates/rbitcoin-rpc/src/methods_tests.rs index 1a0634bed..70a0e1d29 100644 --- a/crates/rbitcoin-rpc/src/methods_tests.rs +++ b/crates/rbitcoin-rpc/src/methods_tests.rs @@ -4275,6 +4275,27 @@ fn getnetworkinfo_includes_p2p_onion() { let _ = std::fs::remove_dir_all(&dir); } +#[test] +fn getnetworkinfo_includes_p2p_i2p() { + use rbitcoin_net::{NetAddr, PeerHub}; + + let (mut ctx, dir) = ctx_empty(); + let hub = PeerHub::new(); + hub.set_discover(false); + let i2p = NetAddr::I2p { + dest: [0x11; 32], + port: 0, + }; + hub.set_p2p_i2p(i2p); + ctx.peers = Some(hub); + let info = dispatch(&ctx, "getnetworkinfo", vec![]).unwrap(); + let addrs = info["localaddresses"].as_array().expect("array"); + assert_eq!(addrs.len(), 1, "{info}"); + assert_eq!(addrs[0]["address"], i2p.host_str()); + assert_eq!(addrs[0]["port"], 0); + let _ = std::fs::remove_dir_all(&dir); +} + #[test] fn getpeerinfo_lists_registered_session() { use bitcoin::p2p::address::Address; diff --git a/crates/rbitcoin-test/Cargo.toml b/crates/rbitcoin-test/Cargo.toml index 5391bf8b8..f1c65f35a 100644 --- a/crates/rbitcoin-test/Cargo.toml +++ b/crates/rbitcoin-test/Cargo.toml @@ -50,5 +50,14 @@ path = "tests/core_analogs.rs" name = "cross_surface" path = "tests/cross_surface.rs" +[features] +# Live private-mesh journeys. Default `cargo test` stays fake SOCKS/SAM/control. +overlay = [] + +[[test]] +name = "overlay" +path = "tests/overlay.rs" +required-features = ["overlay"] + [lints] workspace = true diff --git a/crates/rbitcoin-test/src/lib.rs b/crates/rbitcoin-test/src/lib.rs index 4ad26e58a..c00efebad 100644 --- a/crates/rbitcoin-test/src/lib.rs +++ b/crates/rbitcoin-test/src/lib.rs @@ -4,6 +4,8 @@ pub mod chain_fixture; pub mod mine; +#[cfg(feature = "overlay")] +pub mod overlay; pub use chain_fixture::{ assert_reconstruct_eq, build_mature_regtest_with_spend, pad_empty_from, MatureRegtestChain, diff --git a/crates/rbitcoin-test/src/overlay.rs b/crates/rbitcoin-test/src/overlay.rs new file mode 100644 index 000000000..9f0bf3d9e --- /dev/null +++ b/crates/rbitcoin-test/src/overlay.rs @@ -0,0 +1,470 @@ +//! Helpers for the labeled overlay-functional journeys (real private meshes). + +use bitcoin::hashes::{sha256, Hash}; +use rbitcoin_node::NodeConfig; +use rbitcoin_primitives::Network; +use serde_json::Value; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::net::TcpStream; +use tokio::task::JoinHandle; + +pub const RPC_BEARER: &str = "Bearer pass"; + +pub struct OverlayEnv { + pub tor_socks: SocketAddr, + pub tor_control: SocketAddr, + pub tor_cookie: PathBuf, + pub i2p_sam: SocketAddr, + pub i2p_sam_b: SocketAddr, + pub cjdns_a: std::net::Ipv6Addr, + pub cjdns_b: std::net::Ipv6Addr, +} + +pub fn require_env() -> OverlayEnv { + OverlayEnv { + tor_socks: req_addr("OVERLAY_TOR_SOCKS"), + tor_control: req_addr("OVERLAY_TOR_CONTROL"), + tor_cookie: PathBuf::from(req("OVERLAY_TOR_COOKIE")), + i2p_sam: req_addr("OVERLAY_I2P_SAM"), + i2p_sam_b: req_addr("OVERLAY_I2P_SAM_B"), + cjdns_a: req("OVERLAY_CJDNS_A") + .parse() + .unwrap_or_else(|e| panic!("OVERLAY_CJDNS_A: {e}")), + cjdns_b: req("OVERLAY_CJDNS_B") + .parse() + .unwrap_or_else(|e| panic!("OVERLAY_CJDNS_B: {e}")), + } +} + +fn req(name: &str) -> String { + std::env::var(name) + .unwrap_or_else(|_| panic!("{name} is unset; run via ./scripts/overlay-functional/run.sh")) +} + +fn req_addr(name: &str) -> SocketAddr { + req(name).parse().unwrap_or_else(|e| panic!("{name}: {e}")) +} + +pub async fn live_p2p_lock() -> tokio::sync::MutexGuard<'static, ()> { + static LOCK: std::sync::OnceLock> = std::sync::OnceLock::new(); + LOCK.get_or_init(|| tokio::sync::Mutex::new(())) + .lock() + .await +} + +pub fn ephemeral_addr() -> SocketAddr { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + drop(listener); + addr +} + +pub fn write_rpc_token(datadir: &Path) { + std::fs::write(datadir.join("rpc.token"), "pass").unwrap(); +} + +pub fn base_node(datadir: &Path) -> NodeConfig { + let mut cfg = NodeConfig::default() + .with_datadir(datadir) + .with_network(Network::Regtest) + .with_tiny_heads(); + cfg.listen.use_seeds = false; + cfg.listen.discover = false; + cfg.max_tip_age_secs = Some(u64::MAX); + cfg +} + +pub struct NodeProc { + rpc: SocketAddr, + handle: Option>>, +} + +impl NodeProc { + pub async fn stop(mut self) { + let _ = jsonrpc(self.rpc, "stop", serde_json::json!([])).await; + let handle = self.handle.take().expect("stop once"); + match tokio::time::timeout(Duration::from_secs(15), handle).await { + Ok(Ok(Ok(()))) => {} + Ok(Ok(Err(e))) => panic!("run_p2p: {e}"), + Ok(Err(e)) => panic!("run_p2p join: {e}"), + Err(_) => panic!("run_p2p did not exit after stop"), + } + } +} + +impl Drop for NodeProc { + fn drop(&mut self) { + if let Some(h) = self.handle.take() { + h.abort(); + } + } +} + +pub async fn spawn_node(cfg: NodeConfig, rpc: SocketAddr) -> NodeProc { + // Dedicated blocking pool thread: `run_p2p` does tip-accept on the + // future that `block_on` polls. Tokio names spawn_blocking threads + // `tokio-rt-worker`; BlockingRegion is the same seam as live P2P tests. + let handle = tokio::task::spawn_blocking(move || { + let _block = rbitcoin_net::BlockingRegion::enter(); + tokio::runtime::Handle::current().block_on(rbitcoin_node::run_p2p(cfg)) + }); + let deadline = Instant::now() + Duration::from_secs(120); + loop { + if handle.is_finished() { + match handle.await { + Ok(Ok(())) => panic!("run_p2p exited before rpc {rpc}"), + Ok(Err(e)) => panic!("run_p2p: {e}"), + Err(e) => panic!("run_p2p join: {e}"), + } + } + if TcpStream::connect(rpc).await.is_ok() { + tokio::time::sleep(Duration::from_millis(100)).await; + if handle.is_finished() { + match handle.await { + Ok(Ok(())) => panic!("run_p2p exited after rpc {rpc}"), + Ok(Err(e)) => panic!("run_p2p: {e}"), + Err(e) => panic!("run_p2p join: {e}"), + } + } + return NodeProc { + rpc, + handle: Some(handle), + }; + } + if Instant::now() >= deadline { + panic!("listeners not up ({rpc})"); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +pub async fn jsonrpc(addr: SocketAddr, method: &str, params: Value) -> Value { + jsonrpc_try(addr, method, params) + .await + .unwrap_or_else(|e| panic!("{e}")) +} + +async fn jsonrpc_try(addr: SocketAddr, method: &str, params: Value) -> Result { + let body = serde_json::json!({"jsonrpc":"1.0","id":"test","method":method,"params":params}) + .to_string(); + let req = format!( + "POST / HTTP/1.1\r\nHost: {addr}\r\nAuthorization: {RPC_BEARER}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let mut stream = TcpStream::connect(addr) + .await + .map_err(|e| format!("rpc connect: {e}"))?; + stream + .write_all(req.as_bytes()) + .await + .map_err(|e| format!("rpc write: {e}"))?; + let mut buf = Vec::new(); + stream + .read_to_end(&mut buf) + .await + .map_err(|e| format!("rpc read: {e}"))?; + let text = String::from_utf8_lossy(&buf); + let json_body = text.split("\r\n\r\n").nth(1).unwrap_or("").trim(); + serde_json::from_str(json_body).map_err(|e| format!("rpc {method} json: {e} body={json_body}")) +} + +pub async fn wait_jsonrpc( + addr: SocketAddr, + method: &str, + params: Value, + wall: Duration, + ok: impl Fn(&Value) -> bool, +) -> Value { + let deadline = Instant::now() + wall; + loop { + let last = match jsonrpc_try(addr, method, params.clone()).await { + Ok(v) if ok(&v) => return v, + Ok(v) => v.to_string(), + Err(e) => e, + }; + if Instant::now() >= deadline { + panic!("{method} wall {wall:?}: {last}"); + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +pub fn localaddresses(info: &Value) -> &[Value] { + info["result"]["localaddresses"] + .as_array() + .expect("localaddresses") +} + +pub fn onion_host(row: &Value) -> Option<&str> { + row["address"].as_str().filter(|s| s.ends_with(".onion")) +} + +pub async fn wait_p2p_onion(rpc: SocketAddr, wall: Duration) -> (String, u16) { + let info = wait_jsonrpc(rpc, "getnetworkinfo", serde_json::json!([]), wall, |v| { + localaddresses(v) + .iter() + .any(|r| onion_host(r).is_some() && r["port"] == Network::Regtest.default_p2p_port()) + }) + .await; + for row in localaddresses(&info) { + if let Some(host) = onion_host(row) { + if row["port"] == Network::Regtest.default_p2p_port() { + return (host.to_string(), Network::Regtest.default_p2p_port()); + } + } + } + panic!("p2p onion missing: {info}"); +} + +/// SOCKS domain CONNECT until the onion TCP port accepts (HSDir / ADD_ONION ready). +pub async fn wait_socks_onion(proxy: SocketAddr, host: &str, port: u16, wall: Duration) { + let deadline = Instant::now() + wall; + loop { + let dialer = rbitcoin_net::Dialer::socks(proxy, true); + match dialer.connect_domain(host, port).await { + Ok(_) => return, + Err(e) => { + if Instant::now() >= deadline { + panic!("SOCKS onion {host}:{port} wall {wall:?}: {e}"); + } + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// SAM STREAM CONNECT until the dest accepts (leaseset published on the private mesh). +/// One throwaway session, then drop it before the product node uses this SAM. +pub async fn wait_i2p_stream(sam: SocketAddr, dest_b32: &str, wall: Duration) { + let deadline = Instant::now() + wall; + let session = loop { + match rbitcoin_net::I2pSam::connect(sam).await { + Ok(s) => break s, + Err(e) => { + if Instant::now() >= deadline { + panic!("I2P session {sam} wall {wall:?}: {e}"); + } + } + } + tokio::time::sleep(Duration::from_millis(500)).await; + }; + loop { + match session.stream_connect(dest_b32).await { + Ok(_) => return, + Err(e) => { + if Instant::now() >= deadline { + panic!("I2P STREAM {dest_b32} wall {wall:?}: {e}"); + } + } + } + tokio::time::sleep(Duration::from_millis(500)).await; + } +} + +pub async fn wait_onion_port(rpc: SocketAddr, port: u16, wall: Duration) -> String { + let info = wait_jsonrpc(rpc, "getnetworkinfo", serde_json::json!([]), wall, |v| { + localaddresses(v) + .iter() + .any(|r| onion_host(r).is_some() && r["port"] == port) + }) + .await; + for row in localaddresses(&info) { + if onion_host(row).is_some() && row["port"] == port { + return onion_host(row).unwrap().to_string(); + } + } + panic!("onion port {port} missing: {info}"); +} + +pub async fn wait_peer_network(rpc: SocketAddr, net: &str, inbound: bool, wall: Duration) -> Value { + wait_jsonrpc(rpc, "getpeerinfo", serde_json::json!([]), wall, |v| { + v["result"].as_array().is_some_and(|rows| { + rows.iter().any(|p| { + p["network"] == net && p["inbound"] == inbound && p["startingheight"] != -1 + }) + }) + }) + .await +} + +/// I2P b32 hostname from a SAM `DESTINATION=` / `{datadir}/i2p/p2p.priv` blob. +/// SESSION STATUS returns the private dest; hash is of the public Identity prefix. +pub fn i2p_b32_from_dest(dest: &str) -> String { + let raw = decode_i2p_b64(dest.trim()); + let digest = sha256::Hash::hash(i2p_public_ident(&raw)); + format!("{}.b32.i2p", encode_b32_nopad(digest.as_byte_array())) +} + +fn i2p_public_ident(raw: &[u8]) -> &[u8] { + if raw.len() < 387 { + return raw; + } + let cert_len = u16::from_be_bytes([raw[385], raw[386]]) as usize; + let ident_len = 387usize.saturating_add(cert_len); + if ident_len <= raw.len() { + &raw[..ident_len] + } else { + raw + } +} + +fn decode_i2p_b64(s: &str) -> Vec { + let mut bytes = Vec::new(); + let mut acc = 0u32; + let mut n = 0; + for c in s.chars() { + let v = match c { + 'A'..='Z' => c as u32 - 'A' as u32, + 'a'..='z' => c as u32 - 'a' as u32 + 26, + '0'..='9' => c as u32 - '0' as u32 + 52, + '-' => 62, + '~' => 63, + '+' => 62, + '/' => 63, + '=' => break, + _ => continue, + }; + acc = (acc << 6) | v; + n += 6; + if n >= 8 { + n -= 8; + bytes.push((acc >> n) as u8); + } + } + bytes +} + +fn encode_b32_nopad(data: &[u8]) -> String { + const ALPH: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567"; + let mut out = String::new(); + let mut acc = 0u32; + let mut n = 0; + for &b in data { + acc = (acc << 8) | u32::from(b); + n += 8; + while n >= 5 { + n -= 5; + out.push(ALPH[((acc >> n) & 31) as usize] as char); + } + } + if n > 0 { + out.push(ALPH[((acc << (5 - n)) & 31) as usize] as char); + } + out +} + +pub async fn socks_http_get(proxy: SocketAddr, host: &str, port: u16, path: &str) -> (u16, String) { + let dialer = rbitcoin_net::Dialer::socks(proxy, true); + let mut stream = dialer + .connect_domain(host, port) + .await + .unwrap_or_else(|e| panic!("SOCKS {host}:{port}: {e}")); + let req = format!("GET {path} HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n"); + stream.write_all(req.as_bytes()).await.unwrap(); + let mut buf = Vec::new(); + stream.read_to_end(&mut buf).await.unwrap(); + let text = String::from_utf8_lossy(&buf); + let status = text + .split_whitespace() + .nth(1) + .and_then(|s| s.parse().ok()) + .unwrap_or(0); + let body = text + .split("\r\n\r\n") + .nth(1) + .unwrap_or("") + .trim() + .to_string(); + (status, body) +} + +pub async fn socks_electrum_rpc( + proxy: SocketAddr, + host: &str, + port: u16, + method: &str, + params: Value, +) -> Value { + let dialer = rbitcoin_net::Dialer::socks(proxy, true); + let mut stream = dialer + .connect_domain(host, port) + .await + .unwrap_or_else(|e| panic!("SOCKS electrum {host}:{port}: {e}")); + let req = serde_json::json!({"id":1,"jsonrpc":"2.0","method":method,"params":params}) + .to_string() + + "\n"; + stream.write_all(req.as_bytes()).await.unwrap(); + let mut line = String::new(); + BufReader::new(stream) + .read_line(&mut line) + .await + .expect("electrum read"); + serde_json::from_str(line.trim()) + .unwrap_or_else(|e| panic!("electrum {method}: {e} body={line}")) +} + +pub async fn wait_file(path: &Path, wall: Duration) -> String { + let deadline = Instant::now() + wall; + loop { + if let Ok(s) = std::fs::read_to_string(path) { + let t = s.trim(); + if !t.is_empty() { + return t.to_string(); + } + } + if Instant::now() >= deadline { + panic!("missing {} after {wall:?}", path.display()); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bitcoin::hashes::{sha256, Hash}; + + fn i2p_b64_encode(raw: &[u8]) -> String { + const ALPH: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-~"; + let mut out = String::new(); + let mut i = 0; + while i < raw.len() { + let b0 = raw[i]; + let b1 = if i + 1 < raw.len() { raw[i + 1] } else { 0 }; + let b2 = if i + 2 < raw.len() { raw[i + 2] } else { 0 }; + out.push(ALPH[(b0 >> 2) as usize] as char); + out.push(ALPH[(((b0 & 3) << 4) | (b1 >> 4)) as usize] as char); + if i + 1 < raw.len() { + out.push(ALPH[(((b1 & 15) << 2) | (b2 >> 6)) as usize] as char); + } + if i + 2 < raw.len() { + out.push(ALPH[(b2 & 63) as usize] as char); + } + i += 3; + } + out + } + + #[test] + fn i2p_b32_hashes_identity_prefix() { + let mut ident = vec![7u8; 387]; + ident[384] = 0; + ident[385] = 0; + ident[386] = 0; + let mut privd = ident.clone(); + privd.extend_from_slice(&[9u8; 80]); + let a = i2p_b32_from_dest(&i2p_b64_encode(&ident)); + let b = i2p_b32_from_dest(&i2p_b64_encode(&privd)); + assert_eq!(a, b, "private dest must hash the Identity prefix"); + let full = format!( + "{}.b32.i2p", + encode_b32_nopad(sha256::Hash::hash(&privd).as_byte_array()) + ); + assert_ne!(a, full); + assert!(a.ends_with(".b32.i2p"), "{a}"); + assert_eq!(a.len(), 52 + ".b32.i2p".len(), "{a}"); + } +} diff --git a/crates/rbitcoin-test/tests/overlay.rs b/crates/rbitcoin-test/tests/overlay.rs new file mode 100644 index 000000000..1cab6c392 --- /dev/null +++ b/crates/rbitcoin-test/tests/overlay.rs @@ -0,0 +1,389 @@ +//! Labeled overlay-functional journeys over private Tor / i2pd / cjdns meshes. +//! +//! Default `cargo test` never builds this binary. Run only via +//! `./scripts/overlay-functional/run.sh`. + +use bitcoin::consensus::encode::serialize_hex; +use bitcoin::Amount; +use rbitcoin_net::{NetAddr, OnlyNet}; +use rbitcoin_node::{NodeConfig, P2pListen}; +use rbitcoin_test::mine::spend_anyone_can_spend; +use rbitcoin_test::overlay::{ + base_node, ephemeral_addr, i2p_b32_from_dest, jsonrpc, live_p2p_lock, localaddresses, + require_env, socks_electrum_rpc, socks_http_get, spawn_node, wait_file, wait_i2p_stream, + wait_jsonrpc, wait_onion_port, wait_p2p_onion, wait_peer_network, wait_socks_onion, + write_rpc_token, OverlayEnv, +}; +use rbitcoin_test::TempDir; +use serde_json::json; +use std::net::SocketAddr; +use std::str::FromStr; +use std::time::Duration; + +fn tor_b(env: &OverlayEnv, datadir: &std::path::Path, rpc: SocketAddr) -> NodeConfig { + write_rpc_token(datadir); + let mut cfg = base_node(datadir); + cfg.listen.p2p = P2pListen::Off; + cfg.listen.listen_onion = true; + cfg.tor.control = Some(env.tor_control); + cfg.tor.cookie = Some(env.tor_cookie.clone()); + cfg.rpc.listen = Some(rpc); + cfg +} + +fn tor_a( + env: &OverlayEnv, + datadir: &std::path::Path, + rpc: SocketAddr, + onion: &str, + port: u16, +) -> NodeConfig { + write_rpc_token(datadir); + let mut cfg = base_node(datadir); + cfg.listen.p2p = P2pListen::Off; + cfg.listen.proxy = Some(env.tor_socks); + cfg.listen.only_net = vec![OnlyNet::Onion]; + cfg.listen.connect = vec![format!("{onion}:{port}").parse::().unwrap()]; + cfg.rpc.listen = Some(rpc); + cfg +} + +async fn generate_n(rpc: SocketAddr, n: u32) { + let r = jsonrpc(rpc, "generate", json!([n])).await; + assert!(r["error"].is_null(), "generate: {r}"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn tor_onion_two_node_v2() { + let _live = live_p2p_lock().await; + let env = require_env(); + let b_dir = TempDir::new().unwrap(); + let a_dir = TempDir::new().unwrap(); + let b_rpc = ephemeral_addr(); + let a_rpc = ephemeral_addr(); + + let b = spawn_node(tor_b(&env, b_dir.path(), b_rpc), b_rpc).await; + generate_n(b_rpc, 3).await; + let (onion, port) = wait_p2p_onion(b_rpc, Duration::from_secs(90)).await; + wait_socks_onion(env.tor_socks, &onion, port, Duration::from_secs(90)).await; + + let a = spawn_node(tor_a(&env, a_dir.path(), a_rpc, &onion, port), a_rpc).await; + wait_jsonrpc( + a_rpc, + "getblockcount", + json!([]), + Duration::from_secs(90), + |v| v["result"] == 3, + ) + .await; + let peers = wait_peer_network(a_rpc, "onion", false, Duration::from_secs(30)).await; + let row = &peers["result"].as_array().unwrap()[0]; + assert_eq!(row["network"], "onion", "{peers}"); + assert_eq!(row["inbound"], false, "{peers}"); + assert_eq!(row["transport_protocol_type"], "v2", "{peers}"); + assert!(row["addr"].as_str().unwrap().contains(".onion"), "{peers}"); + assert_ne!(row["addr"], "0.0.0.0", "{peers}"); + + let bin = wait_peer_network(b_rpc, "ipv4", true, Duration::from_secs(30)).await; + let brow = bin["result"] + .as_array() + .unwrap() + .iter() + .find(|p| p["inbound"] == true) + .expect("B inbound"); + assert_eq!(brow["transport_protocol_type"], "v2", "{bin}"); + + a.stop().await; + b.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn tor_wallet_hs_electrum_esplora() { + let _live = live_p2p_lock().await; + let env = require_env(); + let dir = TempDir::new().unwrap(); + let rpc = ephemeral_addr(); + let electrum = ephemeral_addr(); + let esplora = ephemeral_addr(); + + write_rpc_token(dir.path()); + let mut cfg = tor_b(&env, dir.path(), rpc); + cfg.listen.electrum = Some(electrum); + cfg.listen.esplora = Some(rbitcoin_esplora::EsploraListen::Tcp(esplora)); + cfg.shindex = true; + let node = spawn_node(cfg, rpc).await; + generate_n(rpc, 1).await; + + let e_host = wait_onion_port(rpc, electrum.port(), Duration::from_secs(90)).await; + wait_socks_onion( + env.tor_socks, + &e_host, + electrum.port(), + Duration::from_secs(90), + ) + .await; + let s_host = wait_onion_port(rpc, esplora.port(), Duration::from_secs(90)).await; + wait_socks_onion( + env.tor_socks, + &s_host, + esplora.port(), + Duration::from_secs(90), + ) + .await; + + let features = socks_electrum_rpc( + env.tor_socks, + &e_host, + electrum.port(), + "server.features", + json!([]), + ) + .await; + let hosts = &features["result"]["hosts"]; + assert!( + hosts + .get(&e_host) + .is_some_and(|h| h["tcp_port"] == electrum.port()), + "features.hosts: {features}" + ); + assert!( + hosts.get(&e_host).and_then(|h| h.get("ssl_port")).is_none(), + "ssl_port must stay unset: {features}" + ); + + let (st, body) = + socks_http_get(env.tor_socks, &s_host, esplora.port(), "/blocks/tip/height").await; + assert_eq!(st, 200, "esplora GET {body}"); + assert_eq!(body.trim(), "1", "tip height: {body}"); + + node.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn i2p_sam_two_node() { + let _live = live_p2p_lock().await; + let env = require_env(); + let b_dir = TempDir::new().unwrap(); + let a_dir = TempDir::new().unwrap(); + let b_rpc = ephemeral_addr(); + let a_rpc = ephemeral_addr(); + + write_rpc_token(b_dir.path()); + let mut cfg_b = base_node(b_dir.path()); + cfg_b.listen.p2p = P2pListen::Socket(ephemeral_addr()); + cfg_b.listen.i2p_sam = Some(env.i2p_sam_b); + cfg_b.listen.i2p_accept_incoming = true; + cfg_b.listen.only_net = vec![OnlyNet::I2p]; + cfg_b.rpc.listen = Some(b_rpc); + let b = spawn_node(cfg_b, b_rpc).await; + generate_n(b_rpc, 3).await; + + let dest = wait_file( + &b_dir.path().join("i2p").join("p2p.priv"), + Duration::from_secs(180), + ) + .await; + let again = std::fs::read_to_string(b_dir.path().join("i2p").join("p2p.priv")).unwrap(); + assert_eq!(dest, again.trim(), "persistent dest round-trip"); + let b32 = i2p_b32_from_dest(&dest); + let info = jsonrpc(b_rpc, "getnetworkinfo", json!([])).await; + let published = localaddresses(&info) + .iter() + .find(|r| { + r["address"] + .as_str() + .is_some_and(|a| a.ends_with(".b32.i2p")) + }) + .unwrap_or_else(|| panic!("i2p local address missing: {info}")); + assert_eq!(published["address"], b32, "{info}"); + assert_eq!(published["port"], 0, "SAM 3.1 port is 0: {info}"); + let dest_addr: NetAddr = format!("{b32}:0").parse().unwrap(); + wait_i2p_stream(env.i2p_sam, &b32, Duration::from_secs(180)).await; + tokio::time::sleep(Duration::from_secs(2)).await; + + write_rpc_token(a_dir.path()); + let mut cfg_a = base_node(a_dir.path()); + cfg_a.listen.p2p = P2pListen::Off; + cfg_a.listen.i2p_sam = Some(env.i2p_sam); + cfg_a.listen.only_net = vec![OnlyNet::I2p]; + cfg_a.listen.connect = vec![dest_addr]; + cfg_a.rpc.listen = Some(a_rpc); + let a = spawn_node(cfg_a, a_rpc).await; + wait_jsonrpc( + a_rpc, + "getblockcount", + json!([]), + Duration::from_secs(180), + |v| v["result"] == 3, + ) + .await; + let peers = wait_peer_network(a_rpc, "i2p", false, Duration::from_secs(60)).await; + assert_eq!(peers["result"][0]["network"], "i2p", "{peers}"); + assert_eq!( + peers["result"][0]["transport_protocol_type"], "v2", + "{peers}" + ); + let addr = peers["result"][0]["addr"].as_str().unwrap_or(""); + assert!( + addr.starts_with(&b32) && addr.ends_with(":0"), + "dialed peer must be the published dest: {peers}" + ); + let addrv2 = peers["result"][0]["bytesrecv_per_msg"]["addrv2"] + .as_u64() + .unwrap_or(0); + assert!( + addrv2 > 0, + "inbound I2P node must addrv2 its destination: {peers}" + ); + + a.stop().await; + b.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn cjdns_tun_two_node() { + let _live = live_p2p_lock().await; + let env = require_env(); + let b_dir = TempDir::new().unwrap(); + let a_dir = TempDir::new().unwrap(); + let b_rpc = ephemeral_addr(); + let a_rpc = ephemeral_addr(); + + let b_p2p = { + let l = std::net::TcpListener::bind((env.cjdns_b, 0)).unwrap_or_else(|e| { + panic!("bind {} (TUN missing?): {e}", env.cjdns_b); + }); + l.local_addr().unwrap() + }; + let a_p2p = { + let l = std::net::TcpListener::bind((env.cjdns_a, 0)).unwrap_or_else(|e| { + panic!("bind {} (TUN missing?): {e}", env.cjdns_a); + }); + l.local_addr().unwrap() + }; + + write_rpc_token(b_dir.path()); + let mut cfg_b = base_node(b_dir.path()); + cfg_b.listen.p2p = P2pListen::Socket(b_p2p); + cfg_b.listen.cjdns_reachable = true; + cfg_b.listen.discover = true; + cfg_b.listen.only_net = vec![OnlyNet::Cjdns]; + cfg_b.rpc.listen = Some(b_rpc); + let b = spawn_node(cfg_b, b_rpc).await; + generate_n(b_rpc, 3).await; + + write_rpc_token(a_dir.path()); + let mut cfg_a = base_node(a_dir.path()); + cfg_a.listen.p2p = P2pListen::Socket(a_p2p); + cfg_a.listen.cjdns_reachable = true; + cfg_a.listen.discover = true; + cfg_a.listen.only_net = vec![OnlyNet::Cjdns]; + cfg_a.listen.connect = vec![NetAddr::from_str(&b_p2p.to_string()).unwrap()]; + cfg_a.rpc.listen = Some(a_rpc); + let a = spawn_node(cfg_a, a_rpc).await; + wait_jsonrpc( + a_rpc, + "getblockcount", + json!([]), + Duration::from_secs(30), + |v| v["result"] == 3, + ) + .await; + let peers = wait_peer_network(a_rpc, "cjdns", false, Duration::from_secs(20)).await; + assert_eq!(peers["result"][0]["network"], "cjdns", "{peers}"); + assert_eq!( + peers["result"][0]["transport_protocol_type"], "v2", + "{peers}" + ); + let bin = wait_peer_network(b_rpc, "cjdns", true, Duration::from_secs(20)).await; + assert!( + bin["result"] + .as_array() + .unwrap() + .iter() + .any(|p| p["inbound"] == true && p["network"] == "cjdns"), + "{bin}" + ); + let binfo = jsonrpc(b_rpc, "getnetworkinfo", json!([])).await; + assert!( + localaddresses(&binfo) + .iter() + .any(|r| { r["address"] == env.cjdns_b.to_string() && r["port"] == b_p2p.port() }), + "cjdns listen must be a local address: {binfo}" + ); + let ainfo = jsonrpc(a_rpc, "getnetworkinfo", json!([])).await; + assert!( + localaddresses(&ainfo) + .iter() + .any(|r| { r["address"] == env.cjdns_a.to_string() && r["port"] == a_p2p.port() }), + "cjdns listen must be a local address: {ainfo}" + ); + + a.stop().await; + b.stop().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn ephemeral_socks_isolated_broadcast() { + let _live = live_p2p_lock().await; + let env = require_env(); + let b_dir = TempDir::new().unwrap(); + let a_dir = TempDir::new().unwrap(); + let b_rpc = ephemeral_addr(); + let a_rpc = ephemeral_addr(); + + let b = spawn_node(tor_b(&env, b_dir.path(), b_rpc), b_rpc).await; + generate_n(b_rpc, 101).await; + let (onion, port) = wait_p2p_onion(b_rpc, Duration::from_secs(90)).await; + wait_socks_onion(env.tor_socks, &onion, port, Duration::from_secs(90)).await; + + let a = spawn_node(tor_a(&env, a_dir.path(), a_rpc, &onion, port), a_rpc).await; + wait_jsonrpc( + a_rpc, + "getblockcount", + json!([]), + Duration::from_secs(90), + |v| v["result"] == 101, + ) + .await; + wait_peer_network(a_rpc, "onion", false, Duration::from_secs(30)).await; + + let hash = jsonrpc(b_rpc, "getblockhash", json!([1])).await["result"] + .as_str() + .unwrap() + .to_string(); + let blk = jsonrpc(b_rpc, "getblock", json!([hash, 2])).await; + let coinbase = blk["result"]["tx"][0]["txid"].as_str().unwrap(); + let txid = bitcoin::Txid::from_str(coinbase).unwrap(); + let spend = spend_anyone_can_spend(txid, 0, Amount::from_sat(4_999_990_000)); + let hex = serialize_hex(&spend); + let sent = jsonrpc(a_rpc, "sendrawtransaction", json!([hex])).await; + assert!(sent["error"].is_null(), "sendrawtransaction: {sent}"); + let want = spend.compute_txid().to_string(); + + tokio::time::sleep(Duration::from_millis(400)).await; + let standing = jsonrpc(a_rpc, "getpeerinfo", json!([])).await; + for p in standing["result"].as_array().unwrap() { + let inv = p["bytessent_per_msg"] + .get("inv") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + assert_eq!(inv, 0, "standing session must not INV local tx: {standing}"); + } + + wait_jsonrpc( + b_rpc, + "getrawmempool", + json!([]), + Duration::from_secs(90), + |v| { + v["result"] + .as_array() + .is_some_and(|m| m.iter().any(|t| t == &want)) + }, + ) + .await; + + a.stop().await; + b.stop().await; +} diff --git a/docs/ORIENT.md b/docs/ORIENT.md index 6e3e79a88..9f32a33aa 100644 --- a/docs/ORIENT.md +++ b/docs/ORIENT.md @@ -39,6 +39,7 @@ Before the first edit in a crate, read `crates//AGENTS.md` when that file | Open, push, or poll a PR | [`../.agents/skills/ship-pr/SKILL.md`](../.agents/skills/ship-pr/SKILL.md) (one worktree; fail-fast `pr-checks-watch.sh`) | | Minor, patch, or major release | [`../.agents/skills/release/SKILL.md`](../.agents/skills/release/SKILL.md) | | Core functional harness | [`../.agents/skills/core-functional/SKILL.md`](../.agents/skills/core-functional/SKILL.md) | +| Overlay functional harness (private Tor / i2pd / cjdns) | [`../.agents/skills/overlay-functional/SKILL.md`](../.agents/skills/overlay-functional/SKILL.md) | ## Ask first diff --git a/docs/README.md b/docs/README.md index 56067c445..07ef2599d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -46,6 +46,7 @@ into `AGENTS.md`. | [`rpc.md`](./rpc.md) | Core-class JSON-RPC subset. | | [`consensus-tests.md`](./consensus-tests.md) | Rules we own vs Core corpora. | | [`core-functional.md`](./core-functional.md) | Core v31.1 functional harness. | +| [`overlay-functional.md`](./overlay-functional.md) | Private Tor / i2pd / cjdns mesh harness (labeled / nightly). | | [`how-we-plan.md`](./how-we-plan.md) | Red → Green → Refactor planning contract; Agent RAM (redirect cargo/clippy/rustc stdout). | | [`personal-node-plans/`](./personal-node-plans/) | Home-node implementation plan group (SOCKS, overlays, wallet onions, ephemeral broadcast, inwit prune / `NETWORK_LIMITED`). Live flags stay OPERATOR/COMPAT when a slice ships. Do not copy step lists into quality.md until scheduled. | | [`releases.md`](./releases.md) | Tag `vX.Y.Z`, `vX.Y.x` patch line, `.99` bump, Highlights / GitHub notes. | diff --git a/docs/overlay-functional.md b/docs/overlay-functional.md new file mode 100644 index 000000000..e6297a286 --- /dev/null +++ b/docs/overlay-functional.md @@ -0,0 +1,95 @@ +# Overlay functional harness (private meshes) + +Private Tor, i2pd, and cjdns daemons on the runner. Product journeys use the +same operator sockets a home node uses (SOCKS, Tor control cookie, SAM, +`fc00::/8` on a real TUN). This is the **documented exception** to +[personal-node-plans](./personal-node-plans/README.md): product PRs and +default `cargo test` stay fake SOCKS / control / SAM. This job is the +live-daemon oracle. + +Default `cargo test --workspace` never builds the overlay test binary +(`required-features = ["overlay"]`). Do not add `#[ignore]` overlay tests +to the default suite. + +Owner of daemon pins, env vars, walls, and the private-mesh rule. Skill: +[`.agents/skills/overlay-functional/SKILL.md`](../.agents/skills/overlay-functional/SKILL.md). + +## Private mesh (no public overlays) + +GitHub Actions starts: + +| Mesh | How | Must not | +|------|-----|----------| +| Tor | TestingTorNetwork (chutney `basic-min` equivalent: dirauths + client). `AssumeReachable`, short voting. Client SOCKS + control cookie. | Bootstrap public Tor / public HSDir wait | +| i2pd | Six floodfills, **`netid` 16** (not mainnet 2), empty `reseed.urls`, zip-seeded RouterInfos, NTCP2 on `127.0.0.1`, SAM on n0/n1 | Public I2P reseed / netid 2 | +| cjdns | Two `cjdroute` processes, UDP peer on `127.0.0.1`, real TUN (`rbtc0` / `rbtc1`) | Silent bind to `127.0.0.1` if TUN fails | + +Tor SOCKS CONNECT to `127.0.0.1` is unsafe/rejected. Tor journeys are +**onion hidden services** (SOCKS domain CONNECT to `.onion`). Do not “prove +proxy” by CONNECT to loopback. + +## Runner + +```bash +./scripts/overlay-functional/run.sh +./scripts/overlay-functional/run.sh tor_onion_two_node_v2 +./scripts/overlay-functional/run.sh --list +``` + +`run.sh` enters `nix develop .#overlayFunctional` (same rustc 1.95 pin as +the default flake shell, plus `tor` / `i2pd` / `cjdns` from `flake.lock`). +Then: + +1. Start the three private meshes. +2. `cargo build -p rbitcoin-node` +3. `cargo test -p rbitcoin-test --features overlay --test overlay -- --quiet …` + +Timeout **45 minutes** (i2pd tunnel build on a tiny net is the long pole). + +### Env (exported by `run.sh`) + +| Variable | Meaning | +|----------|---------| +| `OVERLAY_TOR_SOCKS` | Client SocksPort `host:port` | +| `OVERLAY_TOR_CONTROL` | ControlPort `host:port` | +| `OVERLAY_TOR_COOKIE` | CookieAuthFile path (SAFECOOKIE) | +| `OVERLAY_I2P_SAM` | Node A SAM (`127.0.0.1:port`) | +| `OVERLAY_I2P_SAM_B` | Node B SAM (accept-incoming) | +| `OVERLAY_CJDNS_A` / `OVERLAY_CJDNS_B` | `fc00::/8` IPv6 on each TUN | + +### Per-journey walls + +| Journey | Wall | +|---------|------| +| Tor HS published (`getnetworkinfo` onion) | ≤ 90s | +| I2P SAM HELLO + STREAM CONNECT | ≤ 180s | +| cjdns TUN TCP | ≤ 60s | + +## Journeys (`tests/overlay.rs`) + +1. **Tor P2P onion** — B `--listen-onion --tor-control`. A `--proxy socks --only-net onion --connect b.onion --no-listen`. BIP324 v2. A `getpeerinfo.network` is `onion`, not `0.0.0.0`. B has inbound. +2. **Tor wallet HS** — B Electrum + Esplora `ADD_ONION`. Client SOCKS domain CONNECT to `.onion`. Pin `server.features.hosts` and one Esplora REST GET. +3. **I2P SAM** — A/B `--i2p-sam --only-net i2p`; B `--i2p-accept-incoming`. B's `getnetworkinfo` address is the `.b32.i2p:0` hash of the public Destination in `p2p.priv` (not a private re-encode of the whole key file). A STREAM CONNECTs to that name. `getpeerinfo` is `i2p` / v2 and has received `addrv2`. +4. **CJDNS TUN** — A/B `--cjdns-reachable` on real `fc00::/8`. `getpeerinfo.network=cjdns` on both sides. `getnetworkinfo.localaddresses` lists each TUN address without `--external-ip`. TUN create failure is a hard fail. +5. **Ephemeral broadcast** — A `--proxy` (real Tor SOCKS); standing onion peer to B; `sendrawtransaction`; standing session must not INV that tx; a one-shot circuit delivers `tx`. + +Out of this harness: 09 inwit-prune; public overlays; replacing NixOS +module dummy `tor`/`i2pd` units ([`nixos-module-runtime`](../nix/tests/nixos-module-runtime.nix) +stays argv / `After=` only). + +## CI + +[`.github/workflows/overlay-functional.yml`](../.github/workflows/overlay-functional.yml): +nightly `42 6 * * *`, `workflow_dispatch`, label **`overlay-functional`**, +label **`release`**, or ship version (same `if:` as Core functional). **Not** +a required check. Do not fold into `release-extra` until the suite is green +for a few nightlies. + +Unlabeled non-ship PRs stay cargo-only ([`TESTING.md`](../TESTING.md)). +Label harness PRs and ship version-bump PRs. Reproduce locally with `run.sh` +until that script passes; do not push-and-wait on the job as the inner loop. + +```bash +./scripts/overlay-functional/run.sh.test.sh +./scripts/overlay-functional/run.sh --list +``` diff --git a/docs/personal-node-plans/README.md b/docs/personal-node-plans/README.md index 1aecd3afa..cf8f25861 100644 --- a/docs/personal-node-plans/README.md +++ b/docs/personal-node-plans/README.md @@ -34,7 +34,9 @@ A node on CGNAT / no forwarded TCP still: - Tor backend is **system `tor`** (SOCKS + control port). Not Arti-in-process. - Fake SOCKS / fake Tor control / fake I2P SAM under `/tmp`. **No** agent-VM - mainnet. **No** live Tor, I2P, or cjdns router in CI. + mainnet. **No** live Tor, I2P, or cjdns router in product PRs or default + `cargo test`. The labeled/nightly overlay job is the live-daemon oracle + ([`overlay-functional.md`](../overlay-functional.md)). - Prefer zero new crates (hand-rolled SOCKS5, control protocol, SAM). Any crate must pass `cargo deny` (musl operator binary). - Operator CLI is kebab (`--only-net`, `--i2p-sam`, `--cjdns-reachable`, diff --git a/docs/releases.md b/docs/releases.md index 21aefd4f9..86646ddb3 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -105,6 +105,7 @@ Cargo.toml). Those PRs run Core functional even without a label. |-------|-----| | `fmt` `deny` `clippy` `ast-grep` `test` `windows` `macos` `coverage` `nixos-module-eval` | Every PR (`ci.yml`) | | `core-functional` | Nightly, `workflow_dispatch`, label **`core-functional`**, label **`release`**, **or** ship version | +| `overlay-functional` | Nightly (`42 6`), `workflow_dispatch`, label **`overlay-functional`**, label **`release`**, **or** ship version. Not required. Not in `release-extra` yet. | | `nixos-module-runtime` | Label **`nixos-module-runtime`**, `workflow_dispatch`, **or** GitHub Release tags (`release.yml`). Not required. | | `release-extra` | Every PR. **Fails** if the PR is ship and `core-functional` is not success | diff --git a/flake.nix b/flake.nix index c64bec5f4..86c1a76f1 100644 --- a/flake.nix +++ b/flake.nix @@ -130,6 +130,37 @@ ''; }; } + // nixpkgs.lib.optionalAttrs (builtins.elem system systems) { + # Private Tor / i2pd / cjdns meshes for scripts/overlay-functional. + # Not the default shell: extra daemons, Linux-only (cjdns TUN). + overlayFunctional = pkgs.mkShell { + packages = with pkgs; [ + rustc + cargo + rustfmt + clippy + llvmPackages.bintools + llvmPackages.llvm + pkg-config + python3 + tor + i2pd + cjdns + iproute2 + procps + iputils + ]; + RUST_BACKTRACE = "1"; + RUSTFLAGS = "-Dwarnings"; + shellHook = '' + export OVERLAY_IN_NIX=1 + if [ -z "''${CARGO_TARGET_DIR:-}" ]; then + export CARGO_TARGET_DIR="$PWD/target/dev" + fi + echo "rbitcoin overlayFunctional: rustc=$(rustc --version) tor=$(tor --version | head -n1) CARGO_TARGET_DIR=$CARGO_TARGET_DIR" + ''; + }; + } ); # `nix flake check` can validate the package builds on the current system. diff --git a/scripts/overlay-functional/cjdns.sh b/scripts/overlay-functional/cjdns.sh new file mode 100755 index 000000000..14fac92ca --- /dev/null +++ b/scripts/overlay-functional/cjdns.sh @@ -0,0 +1,204 @@ +#!/usr/bin/env bash +# Two-node cjdns TUN that UDP-peers on 127.0.0.1. Fail if TUN cannot be created. +set -euo pipefail + +ROOT="${1:?dir}" +mkdir -p "$ROOT" + +if [[ ! -e /dev/net/tun ]]; then + sudo modprobe tun || true +fi +if [[ ! -e /dev/net/tun ]]; then + echo "cjdns: /dev/net/tun missing — no silent 127.0.0.1 fallback" >&2 + exit 1 +fi +if ! sudo -n true 2>/dev/null; then + echo "cjdns: passwordless sudo required for TUN" >&2 + exit 1 +fi + +free_port() { + python3 -c 'import socket; s=socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.bind(("127.0.0.1",0)); print(s.getsockname()[1]); s.close()' +} + +UDP_A="$(free_port)" +UDP_B="$(free_port)" +ADMIN_A="$(free_port)" +ADMIN_B="$(free_port)" + +cjdroute --genconf | cjdroute --cleanconf >"$ROOT/a.raw.json" +cjdroute --genconf | cjdroute --cleanconf >"$ROOT/b.raw.json" + +python3 - "$ROOT" "$UDP_A" "$UDP_B" "$ADMIN_A" "$ADMIN_B" <<'PY' +import json, sys +root, udp_a, udp_b, admin_a, admin_b = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4]), int(sys.argv[5]) + +def load(name): + with open(f"{root}/{name}.raw.json") as f: + return json.load(f) + +a, b = load("a"), load("b") +pw = a["authorizedPasswords"][0]["password"] +a_pub = a["publicKey"] +b_pub = b["publicKey"] + +def udp(cfg): + return cfg["interfaces"]["UDPInterface"][0] + +udp(a)["bind"] = f"0.0.0.0:{udp_a}" +udp(b)["bind"] = f"0.0.0.0:{udp_b}" +udp(a)["connectTo"] = { + f"127.0.0.1:{udp_b}": { + "password": b["authorizedPasswords"][0]["password"], + "publicKey": b_pub, + "peerName": "b", + } +} +udp(b)["connectTo"] = { + f"127.0.0.1:{udp_a}": {"password": pw, "publicKey": a_pub, "peerName": "a"} +} +# One UDP bind (IPv4 loopback). Drop the IPv6 UDPInterface row. +a["interfaces"]["UDPInterface"] = [udp(a)] +b["interfaces"]["UDPInterface"] = [udp(b)] +a["router"]["dnsSeeds"] = [] +b["router"]["dnsSeeds"] = [] +a["security"] = [{"keepNetAdmin": 1}, {"noforks": 1}, {"setupComplete": 1}] +b["security"] = [{"keepNetAdmin": 1}, {"noforks": 1}, {"setupComplete": 1}] +a["logging"] = {"logTo": "stdout"} +b["logging"] = {"logTo": "stdout"} +a["admin"]["bind"] = f"127.0.0.1:{admin_a}" +b["admin"]["bind"] = f"127.0.0.1:{admin_b}" +a["pipe"] = f"{root}/a.cjdns.sock" +b["pipe"] = f"{root}/b.cjdns.sock" + +def tun(cfg, dev): + iface = cfg.setdefault("router", {}).setdefault("interface", {}) + iface["type"] = "TUNInterface" + iface["tunDevice"] = dev + cfg["noBackground"] = 1 + +tun(a, "rbtc0") +tun(b, "rbtc1") + +for name, cfg in (("a", a), ("b", b)): + with open(f"{root}/{name}.conf", "w") as f: + json.dump(cfg, f, indent=2) + f.write("\n") + with open(f"{root}/{name}.ipv6", "w") as f: + f.write(cfg["ipv6"] + "\n") +PY + +CJDNS_BIN="$(command -v cjdroute)" + +# cjdroute reads stdin until EOF as JSON conf, then `--nobg` keeps the +# client in the event loop so the core is not reaped (parent would +# otherwise `return 0`). setsid: cargo/nix must not SIGHUP the pair. +start_cjdroute() { + local name="$1" + sudo bash -c 'exec setsid "$1" --nobg <"$2" >"$3" 2>&1' \ + _ "$CJDNS_BIN" "$ROOT/$name.conf" "$ROOT/$name.log" & + echo $! >"$ROOT/$name.pid" +} + +wait_tun() { + local dev="$1" + local i + for i in $(seq 1 50); do + if ip link show "$dev" >/dev/null 2>&1; then + return 0 + fi + sleep 0.2 + done + echo "cjdns: $dev never appeared" >&2 + return 1 +} + +start_cjdroute a +wait_tun rbtc0 +start_cjdroute b +wait_tun rbtc1 + +IPV6_A="$(cat "$ROOT/a.ipv6")" +IPV6_B="$(cat "$ROOT/b.ipv6")" + +# Kernel TCP over the TUNs (the product path). ICMP through cjdns is not +# reliable on GitHub-hosted runners even when the ifaces are up. +if ! python3 - "$IPV6_A" "$IPV6_B" <<'PY' +import socket, sys, time +a, b = sys.argv[1], sys.argv[2] +end = time.time() + 60 +last = "" +while time.time() < end: + srv = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + srv.bind((b, 0)) + srv.listen(1) + srv.settimeout(1) + port = srv.getsockname()[1] + cli = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + try: + cli.bind((a, 0)) + cli.settimeout(1) + cli.connect((b, port)) + conn, _ = srv.accept() + conn.close() + sys.exit(0) + except OSError as e: + last = str(e) + finally: + cli.close() + except OSError as e: + last = str(e) + finally: + srv.close() + time.sleep(0.4) +print(f"cjdns TCP timeout ({a} -> {b}): {last}", file=sys.stderr) +sys.exit(1) +PY +then + echo "cjdns TCP timeout ($IPV6_A <-> $IPV6_B)" >&2 + for name in a b; do + echo "--- $name pid ---" >&2 + if [[ -f "$ROOT/$name.pid" ]] && kill -0 "$(cat "$ROOT/$name.pid")" 2>/dev/null; then + echo "alive $(cat "$ROOT/$name.pid")" >&2 + else + echo "dead" >&2 + fi + echo "--- $name.log ---" >&2 + tail -n 80 "$ROOT/$name.log" >&2 || true + done + sudo ip -6 addr show rbtc0 >&2 || true + sudo ip -6 addr show rbtc1 >&2 || true + sudo ip -6 route get "$IPV6_B" >&2 || true + exit 1 +fi + +for name in a b; do + if [[ ! -f "$ROOT/$name.pid" ]] || ! kill -0 "$(cat "$ROOT/$name.pid")" 2>/dev/null; then + echo "cjdns: $name not running after TCP wait" >&2 + tail -n 40 "$ROOT/$name.log" >&2 || true + exit 1 + fi +done +# GHA: angel/core can crash ~1s after the first TCP; require the addrs +# still bind after a settle so we do not hand a dead TUN to the journey. +sleep 2 +python3 - "$IPV6_A" "$IPV6_B" <<'PY' +import socket, sys +for ip in sys.argv[1:]: + s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + try: + s.bind((ip, 0)) + except OSError as e: + print(f"cjdns: bind {ip} after settle: {e}", file=sys.stderr) + sys.exit(1) + finally: + s.close() +PY + +cat >"$ROOT/env" <"$dir/i2pd.conf" <"$ROOT/loopbacks" +for i in $(seq 0 $((N - 1))); do + dir="$ROOT/n$i" + ntcp="$(free_port)" + sam="$(free_port)" + ipv4="127.0.0.$((i + 2))" + NTCPS+=("$ntcp") + SAMS+=("$sam") + echo "$ipv4" >>"$ROOT/loopbacks" + sudo ip addr add "$ipv4/8" dev lo 2>/dev/null || true + write_conf "$dir" "$ntcp" "$sam" "$ipv4" + i2pd --datadir "$dir" --conf "$dir/i2pd.conf" --address4 "$ipv4" --daemon +done + +deadline=$((SECONDS + 30)) +for i in $(seq 0 $((N - 1))); do + while [[ ! -f "$ROOT/n$i/router.info" ]]; do + if (( SECONDS >= deadline )); then + echo "i2pd n$i router.info missing" >&2 + exit 1 + fi + sleep 0.2 + done +done + +python3 - "$ROOT" "$N" <<'PY' +import sys, zipfile +from pathlib import Path +root = Path(sys.argv[1]) +n = int(sys.argv[2]) +zpath = root / "reseed.zip" +with zipfile.ZipFile(zpath, "w", compression=zipfile.ZIP_STORED) as z: + for i in range(n): + ri = root / f"n{i}" / "router.info" + z.write(ri, arcname=f"routerInfo-n{i}.dat") +print(zpath) +PY + +for i in $(seq 0 $((N - 1))); do + if [[ -f "$ROOT/n$i/i2pd.pid" ]]; then + kill "$(cat "$ROOT/n$i/i2pd.pid")" 2>/dev/null || true + fi +done +sleep 1 +for i in $(seq 0 $((N - 1))); do + ipv4="127.0.0.$((i + 2))" + i2pd --datadir "$ROOT/n$i" --conf "$ROOT/n$i/i2pd.conf" \ + --address4 "$ipv4" --reseed.zipfile "$ROOT/reseed.zip" --daemon +done +sleep 3 + +sam_hello() { + local port="$1" wall="$2" + python3 - "$port" "$wall" <<'PY' +import socket, sys, time +port = int(sys.argv[1]) +wall = int(sys.argv[2]) +end = time.time() + wall +last = "" +while time.time() < end: + try: + s = socket.create_connection(("127.0.0.1", port), 2) + s.sendall(b"HELLO VERSION MIN=3.1 MAX=3.1\n") + last = s.recv(1024).decode("utf-8", "replace") + s.close() + if "RESULT=OK" in last.upper(): + sys.exit(0) + except OSError as e: + last = str(e) + time.sleep(0.4) +print(f"SAM {port} not ready: {last}", file=sys.stderr) +sys.exit(1) +PY +} + +sam_hello "${SAMS[0]}" 180 +sam_hello "${SAMS[1]}" 180 +# HELLO can succeed before SESSION CREATE is stable on a tiny net. +sleep 2 + +cat >"$ROOT/env" </dev/null 2>&1; then + echo "overlay-functional: nix is required (flake overlayFunctional shell)" >&2 + exit 1 + fi + exec nix develop "$ROOT#overlayFunctional" --command \ + env OVERLAY_IN_NIX=1 \ + CARGO_TARGET_DIR="${CARGO_TARGET_DIR:-$PWD/target/dev}" \ + "$0" "$@" +fi + +need() { + if ! command -v "$1" >/dev/null 2>&1; then + echo "overlay-functional: missing $1 (use flake overlayFunctional)" >&2 + exit 1 + fi +} +need tor +need tor-gencert +need i2pd +need cjdroute +need python3 +need cargo + +WORKDIR="$(mktemp -d /tmp/rbitcoin-overlay.XXXXXX)" +cleanup() { + local st=$? + set +e + if [[ -d "$WORKDIR/tor" ]]; then + find "$WORKDIR/tor" -name tor.pid -print0 2>/dev/null \ + | xargs -0 -r -I{} sh -c 'kill "$(cat "$1")" 2>/dev/null' _ {} + fi + if [[ -d "$WORKDIR/i2p" ]]; then + for p in "$WORKDIR/i2p"/n*/i2pd.pid; do + [[ -f "$p" ]] && kill "$(cat "$p")" 2>/dev/null + done + if [[ -f "$WORKDIR/i2p/loopbacks" ]]; then + while read -r ip; do + sudo ip addr del "$ip/8" dev lo 2>/dev/null || true + done <"$WORKDIR/i2p/loopbacks" + fi + fi + if [[ -d "$WORKDIR/cjdns" ]]; then + for p in "$WORKDIR/cjdns/a.pid" "$WORKDIR/cjdns/b.pid"; do + [[ -f "$p" ]] && sudo kill "$(cat "$p")" 2>/dev/null + done + sudo pkill -f "cjdroute core $WORKDIR/cjdns" 2>/dev/null || true + sudo ip link del rbtc0 2>/dev/null + sudo ip link del rbtc1 2>/dev/null + fi + rm -rf "$WORKDIR" + exit "$st" +} +trap cleanup EXIT INT TERM + +echo "overlay-functional: workdir $WORKDIR" >&2 + +python3 "$HERE/tornet.py" "$WORKDIR/tor" +# shellcheck disable=SC1091 +source "$WORKDIR/tor/env" +chmod +x "$HERE/i2pnet.sh" "$HERE/cjdns.sh" +"$HERE/i2pnet.sh" "$WORKDIR/i2p" +# shellcheck disable=SC1091 +source "$WORKDIR/i2p/env" +"$HERE/cjdns.sh" "$WORKDIR/cjdns" +# shellcheck disable=SC1091 +source "$WORKDIR/cjdns/env" + +export OVERLAY_TOR_SOCKS OVERLAY_TOR_CONTROL OVERLAY_TOR_COOKIE +export OVERLAY_I2P_SAM OVERLAY_I2P_SAM_B +export OVERLAY_CJDNS_A OVERLAY_CJDNS_B + +echo "overlay-functional: TOR_SOCKS=$OVERLAY_TOR_SOCKS TOR_CONTROL=$OVERLAY_TOR_CONTROL" >&2 +echo "overlay-functional: I2P_SAM=$OVERLAY_I2P_SAM I2P_SAM_B=$OVERLAY_I2P_SAM_B" >&2 +echo "overlay-functional: CJDNS_A=$OVERLAY_CJDNS_A CJDNS_B=$OVERLAY_CJDNS_B" >&2 + +cargo build -p rbitcoin-node +cargo test -p rbitcoin-test --features overlay --test overlay -- --quiet --test-threads=1 "$@" +echo "overlay-functional: journeys done" diff --git a/scripts/overlay-functional/run.sh.test.sh b/scripts/overlay-functional/run.sh.test.sh new file mode 100755 index 000000000..b9c22428a --- /dev/null +++ b/scripts/overlay-functional/run.sh.test.sh @@ -0,0 +1,49 @@ +#!/usr/bin/env bash +# Contract pin for overlay-functional/run.sh (no daemons, no cargo). +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/../.." && pwd)" +RUN="$ROOT/scripts/overlay-functional/run.sh" +PASS=0 +FAIL=0 + +assert_ok() { + local name="$1" + shift + if "$@" >/dev/null 2>&1; then + echo "ok - $name" + PASS=$((PASS + 1)) + else + echo "not ok - $name" + FAIL=$((FAIL + 1)) + fi +} + +assert_stdout() { + local name="$1" + local needle="$2" + shift 2 + local out + if ! out="$("$@" 2>/dev/null)"; then + echo "not ok - $name (unexpected failure)" + FAIL=$((FAIL + 1)) + return + fi + if printf '%s' "$out" | grep -q -- "$needle"; then + echo "ok - $name" + PASS=$((PASS + 1)) + else + echo "not ok - $name (missing '$needle' in stdout: $out)" + FAIL=$((FAIL + 1)) + fi +} + +assert_stdout tor_list tor_onion_two_node_v2 "$RUN" --list +assert_stdout wallet_list tor_wallet_hs_electrum_esplora "$RUN" --list +assert_stdout i2p_list i2p_sam_two_node "$RUN" --list +assert_stdout cjdns_list cjdns_tun_two_node "$RUN" --list +assert_stdout eph_list ephemeral_socks_isolated_broadcast "$RUN" --list +assert_ok help "$RUN" --help + +echo "$PASS passed, $FAIL failed" +[[ "$FAIL" -eq 0 ]] diff --git a/scripts/overlay-functional/tornet.py b/scripts/overlay-functional/tornet.py new file mode 100755 index 000000000..b4b02d45d --- /dev/null +++ b/scripts/overlay-functional/tornet.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Private TestingTorNetwork (chutney basic-min equivalent). Never bootstraps public Tor.""" + +from __future__ import annotations + +import os +import shutil +import socket +import subprocess +import sys +import time +from pathlib import Path + +AUTH_COUNT = 3 +VOTE = 10 + + +def free_port() -> int: + s = socket.socket() + s.bind(("127.0.0.1", 0)) + p = s.getsockname()[1] + s.close() + return p + + +def run(cmd: list[str], **kw) -> subprocess.CompletedProcess: + return subprocess.run(cmd, check=True, text=True, **kw) + + +def write(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + + +def tor_gencert(datadir: Path, dirport: int) -> str: + keys = datadir / "keys" + keys.mkdir(parents=True, exist_ok=True) + ident = keys / "authority_identity_key" + signing = keys / "authority_signing_key" + cert = keys / "authority_certificate" + run( + [ + "tor-gencert", + "--create-identity-key", + "--passphrase-fd", + "0", + "-m", + "12", + "-a", + f"127.0.0.1:{dirport}", + "-i", + str(ident), + "-s", + str(signing), + "-c", + str(cert), + ], + input="\n", + ) + v3ident = None + for line in cert.read_text().splitlines(): + if line.lower().startswith("fingerprint"): + v3ident = line.split()[-1].replace(" ", "") + break + if not v3ident: + raise SystemExit(f"no v3ident in {cert}") + return v3ident + + +def list_fingerprint(datadir: Path, nick: str) -> str: + dummy = datadir / "fp.torrc" + write( + dummy, + "\n".join( + [ + f"DataDirectory {datadir}", + f"Nickname {nick}", + "ORPort 127.0.0.1:1", + "DirPort 0", + "SocksPort 0", + "TestingTorNetwork 1", + 'DirAuthority x orport=1 v3ident=0000000000000000000000000000000000000000 127.0.0.1:1 0000000000000000000000000000000000000000', + ] + ) + + "\n", + ) + out = run( + ["tor", "--quiet", "-f", str(dummy), "--list-fingerprint"], + capture_output=True, + ).stdout.strip() + dummy.unlink(missing_ok=True) + parts = out.split() + fp = "".join(parts[1:]) if len(parts) > 1 else parts[-1] + return fp.replace(" ", "") + + +def dirauth_lines(auths: list[dict]) -> str: + lines = [] + for a in auths: + lines.append( + f"DirAuthority {a['nick']} orport={a['orport']} v3ident={a['v3ident']} " + f"127.0.0.1:{a['dirport']} {a['fp']}" + ) + return "\n".join(lines) + + +def common_torrc(datadir: Path, nick: str, auths: str) -> str: + return "\n".join( + [ + "TestingTorNetwork 1", + "AssumeReachable 1", + "TestingAuthDirTimeToLearnReachability 0", + "TestingMinExitFlagThreshold 0", + f"TestingV3AuthInitialVotingInterval {VOTE}", + "TestingV3AuthInitialVoteDelay 2", + "TestingV3AuthInitialDistDelay 2", + # TestingTorNetwork's ongoing vote is 5 minutes; keep the tiny-net cadence. + f"V3AuthVotingInterval {VOTE}", + "V3AuthVoteDelay 2", + "V3AuthDistDelay 2", + "TestingDirAuthVoteHSDir *", + "TestingDirAuthVoteGuard *", + "TestingDirAuthVoteExit *", + "MinUptimeHidServDirectoryV2 0", + "ConfluxEnabled 0", + "LearnCircuitBuildTimeout 0", + "CircuitBuildTimeout 10", + "PathsNeededToBuildCircuits 0.25", + "DataDirectory {0}".format(datadir), + f"Nickname {nick}", + f"PidFile {datadir / 'tor.pid'}", + f"Log notice file {datadir / 'notice.log'}", + "SafeLogging 0", + "ProtocolWarnings 1", + "ShutdownWaitLength 0", + "DisableDebuggerAttachment 0", + "FetchHidServDescriptors 1", + "HiddenServiceStatistics 0", + "ServerDNSDetectHijacking 0", + "ServerDNSTestAddresses", + auths, + "RunAsDaemon 1", + ] + ) + + +def start_tor(torrc: Path) -> None: + run(["tor", "-f", str(torrc)]) + + +def wait_log(log: Path, needles: tuple[str, ...], seconds: int) -> None: + deadline = time.time() + seconds + while time.time() < deadline: + if log.is_file(): + text = log.read_text(errors="replace") + if any(n in text for n in needles): + return + time.sleep(0.4) + extra = "" + if log.is_file(): + extra = "\n".join(log.read_text(errors="replace").splitlines()[-25:]) + raise SystemExit(f"bootstrap timeout: {log}\n{extra}") + + +def main() -> int: + if len(sys.argv) < 2: + print("usage: tornet.py DIR", file=sys.stderr) + return 2 + root = Path(sys.argv[1]) + if root.exists(): + shutil.rmtree(root) + root.mkdir(parents=True) + env_path = root / "env" + + auths = [] + for i in range(AUTH_COUNT): + nick = f"auth{i}" + datadir = root / nick + datadir.mkdir() + orport = free_port() + dirport = free_port() + v3ident = tor_gencert(datadir, dirport) + fp = list_fingerprint(datadir, nick) + auths.append( + { + "nick": nick, + "datadir": datadir, + "orport": orport, + "dirport": dirport, + "v3ident": v3ident, + "fp": fp, + } + ) + + da = dirauth_lines(auths) + for a in auths: + extra = "\n".join( + [ + "SocksPort 0", + f"ORPort {a['orport']} IPv4Only", + f"DirPort {a['dirport']} IPv4Only", + "Address 127.0.0.1", + "AuthoritativeDirectory 1", + "V3AuthoritativeDirectory 1", + "ContactInfo auth@test.test", + "ExitRelay 1", + "ExitPolicy accept *:*", + "IPv6Exit 0", + ] + ) + torrc = a["datadir"] / "torrc" + write(torrc, common_torrc(a["datadir"], a["nick"], da) + "\n" + extra + "\n") + start_tor(torrc) + + relays = [] + for i in range(3): + nick = f"relay{i}" + datadir = root / nick + datadir.mkdir() + orport = free_port() + fp = list_fingerprint(datadir, nick) + relays.append({"nick": nick, "datadir": datadir, "orport": orport, "fp": fp}) + extra = "\n".join( + [ + "SocksPort 0", + f"ORPort {orport} IPv4Only", + "DirPort 0", + "Address 127.0.0.1", + "ExitRelay 1", + "ExitPolicy accept *:*", + ] + ) + torrc = datadir / "torrc" + write(torrc, common_torrc(datadir, nick, da) + "\n" + extra + "\n") + start_tor(torrc) + + for a in auths: + wait_log( + a["datadir"] / "notice.log", + ("Bootstrapped 100%", "Published ns consensus"), + 90, + ) + for r in relays: + wait_log(r["datadir"] / "notice.log", ("Bootstrapped 100%",), 90) + + client = root / "client" + client.mkdir() + socks = free_port() + control = free_port() + cookie = client / "control_auth_cookie" + extra = "\n".join( + [ + f"SocksPort 127.0.0.1:{socks}", + f"ControlPort 127.0.0.1:{control}", + "CookieAuthentication 1", + f"CookieAuthFile {cookie}", + "ClientOnly 1", + "ORPort 0", + "DirPort 0", + "SocksPolicy accept 127.0.0.0/8", + ] + ) + torrc = client / "torrc" + write(torrc, common_torrc(client, "client", da) + "\n" + extra + "\n") + start_tor(torrc) + wait_log(client / "notice.log", ("Bootstrapped 100%",), 120) + + if not cookie.is_file(): + raise SystemExit("control cookie missing") + write( + env_path, + "\n".join( + [ + f"OVERLAY_TOR_SOCKS=127.0.0.1:{socks}", + f"OVERLAY_TOR_CONTROL=127.0.0.1:{control}", + f"OVERLAY_TOR_COOKIE={cookie}", + ] + ) + + "\n", + ) + print(env_path.read_text(), end="") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())