test(maestro): cover perimeter-mismatch boundary + broker error paths (NAP-270)
ci / rust (push) Failing after 5m25s
ci / rust (push) Failing after 5m25s
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) <noreply@anthropic.com>
This commit is contained in:
@@ -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"] }
|
||||
|
||||
@@ -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:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<maestro_protocol::Session> = 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
|
||||
|
||||
@@ -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<DaemonInbound>,
|
||||
) -> RpcResult<AttachCandidates> {
|
||||
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)]
|
||||
|
||||
Reference in New Issue
Block a user