From f3f696790cf6dabaca95ff7599b469fb7f8f17bd Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Mon, 1 Jun 2026 00:04:16 +0000 Subject: [PATCH] test(maestro): cover perimeter-mismatch boundary + broker error paths (NAP-270) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The asymmetric-VLAN security boundary and the broker's RPC error branches had no tests, so a regression that leaked a JWT across perimeters or dialed across them would ship undetected. Adds the required coverage: registry (state.rs unit): - perimeter_for_requires_both_id_and_addr - resolve_attach_refuses_self_target - resolve_attach_returns_unknown_session_for_orphan_id daemon broker (broker.rs unit): - broker_resolve_attach_propagates_registry_err - broker_resolve_attach_times_out_when_no_reply (paused-clock, instant) daemon e2e: - broker_attach_returns_502_on_perimeter_mismatch (two daemons, different perimeters → structured 502, never dials across) Enables tokio `test-util` in maestro-daemon dev-deps for the paused-clock test. Unblocks NAP-272. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/maestro-daemon/Cargo.toml | 2 + crates/maestro-daemon/src/broker.rs | 56 ++++++++++++++ crates/maestro-e2e/tests/e2e.rs | 107 +++++++++++++++++++++++++++ crates/maestro-registry/src/state.rs | 83 +++++++++++++++++++++ 4 files changed, 248 insertions(+) diff --git a/crates/maestro-daemon/Cargo.toml b/crates/maestro-daemon/Cargo.toml index b5f0338..35f21a2 100644 --- a/crates/maestro-daemon/Cargo.toml +++ b/crates/maestro-daemon/Cargo.toml @@ -35,3 +35,5 @@ uuid.workspace = true reqwest.workspace = true tempfile.workspace = true maestro-registry = { path = "../maestro-registry" } +# `test-util` enables `start_paused` for instant timeout tests (NAP-270). +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs index 6ca94f3..073669e 100644 --- a/crates/maestro-daemon/src/broker.rs +++ b/crates/maestro-daemon/src/broker.rs @@ -204,4 +204,60 @@ mod tests { // It must not leave a pending entry behind either. assert_eq!(handle.pending.lock().unwrap().len(), 0); } + + /// NAP-270: a registry-reported error in the `ResolveAttachResult` reply is + /// surfaced as `BrokerError::Registry` carrying the registry's error code, + /// so the broker handler can map it to the right HTTP status. + #[tokio::test] + async fn broker_resolve_attach_propagates_registry_err() { + let (handle, mut out_rx) = BrokerHandle::new(); + handle.set_connected(true); + + let h = handle.clone(); + let task = tokio::spawn(async move { h.resolve_attach(Uuid::new_v4()).await }); + + // Learn the reply_to from the outbound frame, then deliver an error. + let reply_to = match out_rx.recv().await.expect("ResolveAttach went out") { + DaemonOutbound::ResolveAttach { reply_to, .. } => reply_to, + other => panic!("expected ResolveAttach, got {other:?}"), + }; + handle + .complete_resolve( + reply_to, + RpcResult::Err { + error: maestro_protocol::ErrorInfo { + code: "unknown_session".into(), + message: "no daemon owns it".into(), + }, + }, + ) + .await; + + match task.await.unwrap() { + Err(BrokerError::Registry(code, _)) => assert_eq!(code, "unknown_session"), + other => panic!("expected Registry error, got {other:?}"), + } + } + + /// NAP-270: if the registry never replies, `resolve_attach` resolves to + /// `BrokerError::Timeout` after `RESOLVE_TIMEOUT` (driven here with a paused + /// clock so the test is instant). + #[tokio::test(start_paused = true)] + async fn broker_resolve_attach_times_out_when_no_reply() { + let (handle, mut out_rx) = BrokerHandle::new(); + handle.set_connected(true); + + let h = handle.clone(); + let task = tokio::spawn(async move { h.resolve_attach(Uuid::new_v4()).await }); + + // The request is out and the call is parked on the reply oneshot. + let _ = out_rx.recv().await.expect("ResolveAttach went out"); + // Never reply; jump past the RPC timeout. + tokio::time::advance(RESOLVE_TIMEOUT + Duration::from_secs(1)).await; + + match task.await.unwrap() { + Err(BrokerError::Timeout(_)) => {} + other => panic!("expected Timeout, got {other:?}"), + } + } } diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 310f9a0..7d418b9 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -638,6 +638,113 @@ async fn broker_rejects_when_target_doesnt_allow_inbound() { drop(registry); } +#[tokio::test(flavor = "multi_thread")] +async fn broker_attach_returns_502_on_perimeter_mismatch() { + // NAP-270: the asymmetric-VLAN boundary. alpha and beta are on DIFFERENT + // perimeters, so alpha's broker must refuse T2 (the only implemented + // transport) with a structured 502 rather than dial across perimeters. + let registry = Registry::spawn("e2e-shared-token").await; + let beta_bind = std::net::SocketAddr::from(( + std::net::Ipv4Addr::LOCALHOST, + maestro_e2e::util::free_port(), + )); + let beta = Daemon::spawn_at( + "beta", + ®istry.http_base(), + "e2e-shared-token", + beta_bind, + DaemonOptions { + perimeter_id: Some("wg-beta".into()), + private_addr: Some(beta_bind.to_string()), + allow_inbound: vec!["alpha".into()], + }, + ) + .await; + let alpha_bind = std::net::SocketAddr::from(( + std::net::Ipv4Addr::LOCALHOST, + maestro_e2e::util::free_port(), + )); + let alpha = Daemon::spawn_at( + "alpha", + ®istry.http_base(), + "e2e-shared-token", + alpha_bind, + DaemonOptions { + // Different perimeter from beta → T2 unavailable. + perimeter_id: Some("wg-alpha".into()), + private_addr: Some(alpha_bind.to_string()), + allow_inbound: vec![], + }, + ) + .await; + + let client = TestClient::new(®istry.http_base(), ®istry.ws_base()); + client.bootstrap().await.expect("bootstrap"); + wait_for_async("both daemons connected", Duration::from_secs(10), || { + let client = client.clone(); + async move { + client + .list_daemons() + .await + .map(|ds| ds.len() >= 2) + .unwrap_or(false) + } + }) + .await; + + let cwd = TempDir::new().unwrap(); + let session = client + .spawn_session(&create_req("beta", cwd.path(), "sleep 30")) + .await + .expect("spawn beta"); + let sid = session.id; + + // Wait until alpha's catalog mirror sees beta's session. + let alpha_base = format!("http://{}", alpha.bind_addr); + wait_for_async( + "alpha mirrors beta's session", + Duration::from_secs(5), + || { + let base = alpha_base.clone(); + async move { + let resp = match reqwest::get(format!("{base}/api/sessions")).await { + Ok(r) => r, + Err(_) => return false, + }; + let list: Vec = match resp.json().await { + Ok(v) => v, + Err(_) => return false, + }; + list.iter().any(|s| s.id == sid) + } + }, + ) + .await; + + // Attach through alpha's broker — must 502 (perimeter mismatch), never + // upgrade and never dial beta. + let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid); + let err = tokio_tungstenite::connect_async(broker_url) + .await + .expect_err("cross-perimeter attach must fail the handshake"); + match err { + tokio_tungstenite::tungstenite::Error::Http(resp) => { + assert_eq!( + resp.status(), + 502, + "expected BAD_GATEWAY for perimeter mismatch, got {}", + resp.status() + ); + } + other => panic!("expected an HTTP 502 response, got {other:?}"), + } + + client.kill_session(sid).await.expect("kill"); + drop(alpha); + drop(beta); + drop(registry); +} + #[tokio::test(flavor = "multi_thread")] async fn daemon_events_ws_streams_snapshot_then_session_added() { // NAP-248: subscribe to a daemon's WS /api/events, expect an initial diff --git a/crates/maestro-registry/src/state.rs b/crates/maestro-registry/src/state.rs index 25f25af..8f88bda 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -701,6 +701,89 @@ mod tests { reg.deregister("ghost", Uuid::new_v4()).await; assert!(reg.daemons().await.is_empty()); } + + // ----- NAP-270: perimeter / resolve_attach security boundary ----- + + /// Read the single `ResolveAttachResult` the registry pushes to the home + /// daemon's outbound channel in response to a `resolve_attach`. + async fn recv_resolve_result( + rx: &mut mpsc::Receiver, + ) -> RpcResult { + match rx + .recv() + .await + .expect("home daemon got a ResolveAttachResult") + { + DaemonInbound::ResolveAttachResult { result, .. } => result, + other => panic!("expected ResolveAttachResult, got {other:?}"), + } + } + + /// The `(perimeter_id, private_addr)` pair only yields a candidate when + /// BOTH halves are advertised — a half-configured daemon must not leak a + /// T2 dial target (NAP-270). + #[tokio::test] + async fn perimeter_for_requires_both_id_and_addr() { + let reg = Registry::new(); + let (_t, _rx) = register(®, "d1").await; + + assert!( + reg.perimeter_for("d1").await.is_none(), + "nothing advertised" + ); + + reg.update_perimeter("d1", Some("wg".into()), None).await; + assert!( + reg.perimeter_for("d1").await.is_none(), + "addr missing → None" + ); + + reg.update_perimeter("d1", None, Some("10.0.0.1:7070".into())) + .await; + assert!(reg.perimeter_for("d1").await.is_none(), "id missing → None"); + + reg.update_perimeter("d1", Some("wg".into()), Some("10.0.0.1:7070".into())) + .await; + assert_eq!( + reg.perimeter_for("d1").await, + Some(("wg".into(), "10.0.0.1:7070".into())) + ); + } + + /// A home daemon asking to attach to a session it already owns is refused + /// (`self_target`) rather than handed a useless pair JWT (NAP-270). + #[tokio::test] + async fn resolve_attach_refuses_self_target() { + let reg = Registry::new(); + let (_t, mut home_rx) = register(®, "laptop").await; + let sid = Uuid::new_v4(); + reg.add_session("laptop", sample_session(sid, "laptop")) + .await; + + reg.resolve_attach("laptop", Uuid::new_v4(), sid, "secret") + .await; + + match recv_resolve_result(&mut home_rx).await { + RpcResult::Err { error } => assert_eq!(error.code, "self_target"), + RpcResult::Ok { .. } => panic!("self-target attach must be refused"), + } + } + + /// Resolving an attach for a session no daemon owns yields + /// `unknown_session`, not a JWT for some default daemon (NAP-270). + #[tokio::test] + async fn resolve_attach_returns_unknown_session_for_orphan_id() { + let reg = Registry::new(); + let (_t, mut home_rx) = register(®, "laptop").await; + + reg.resolve_attach("laptop", Uuid::new_v4(), Uuid::new_v4(), "secret") + .await; + + match recv_resolve_result(&mut home_rx).await { + RpcResult::Err { error } => assert_eq!(error.code, "unknown_session"), + RpcResult::Ok { .. } => panic!("orphan session must not resolve"), + } + } } #[cfg(test)]