diff --git a/crates/maestro-e2e/src/daemon.rs b/crates/maestro-e2e/src/daemon.rs index 095ffe5..6ca2943 100644 --- a/crates/maestro-e2e/src/daemon.rs +++ b/crates/maestro-e2e/src/daemon.rs @@ -27,14 +27,52 @@ pub struct Daemon { token: String, } +/// Optional knobs for `Daemon::spawn_with` — perimeter + allow_inbound for +/// NAP-247 broker tests. +#[derive(Debug, Default, Clone)] +pub struct DaemonOptions { + pub perimeter_id: Option, + pub private_addr: Option, + pub allow_inbound: Vec, +} + impl Daemon { /// Spawn a fresh daemon pointing at `registry_base_url` (the registry's /// HTTP base, e.g. `http://127.0.0.1:1234`). The daemon derives its WS /// endpoint internally via `derive_daemon_ws_url`. pub async fn spawn(daemon_id: &str, registry_base_url: &str, token: &str) -> Self { + Self::spawn_with( + daemon_id, + registry_base_url, + token, + DaemonOptions::default(), + ) + .await + } + + /// Spawn with NAP-247 options (perimeter advertisement, allow_inbound). + pub async fn spawn_with( + daemon_id: &str, + registry_base_url: &str, + token: &str, + opts: DaemonOptions, + ) -> Self { let port = free_port(); let bind = SocketAddr::from((Ipv4Addr::LOCALHOST, port)); - Self::spawn_with_port(daemon_id, registry_base_url, token, bind).await + Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await + } + + /// Spawn on a caller-supplied bind address. Used by NAP-247 broker tests + /// where the daemon's advertised `private_addr` must equal the actual + /// bind address so the broker can dial it back. + pub async fn spawn_at( + daemon_id: &str, + registry_base_url: &str, + token: &str, + bind: SocketAddr, + opts: DaemonOptions, + ) -> Self { + Self::spawn_with_port_and_opts(daemon_id, registry_base_url, token, bind, opts).await } /// Stop the underlying process, but keep its state dirs around so we can @@ -48,7 +86,8 @@ impl Daemon { /// Re-spawn the daemon on the same port + state. The previous tmux /// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt - /// its sessions. + /// its sessions. Restarts inherit the original options via the on-disk + /// config file written at `spawn_with_port_and_opts` time. pub async fn restart(&mut self) { self.stop(); let child = launch( @@ -60,21 +99,28 @@ impl Daemon { self.home.path(), self.state.path(), self.tmux.path(), + &DaemonOptions::default(), ); self.child = Some(child); self.wait_until_attach_ready().await; } - async fn spawn_with_port( + async fn spawn_with_port_and_opts( daemon_id: &str, registry_url: &str, token: &str, bind: SocketAddr, + opts: DaemonOptions, ) -> Self { let home = TempDir::new().expect("home tmpdir"); let state = TempDir::new().expect("state tmpdir"); let tmux = TempDir::new().expect("tmux tmpdir"); + // Write a minimal config.toml so allow_inbound (and any future + // file-only fields) reach the daemon. Required fields keep coming + // through env vars set by `launch`. + write_config(home.path(), &opts); + let daemon_url = format!("ws://{bind}"); let child = launch( daemon_id, @@ -85,6 +131,7 @@ impl Daemon { home.path(), state.path(), tmux.path(), + &opts, ); let me = Self { daemon_id: daemon_id.to_string(), @@ -140,10 +187,11 @@ fn launch( home: &std::path::Path, _state: &std::path::Path, tmux_tmpdir: &std::path::Path, + opts: &DaemonOptions, ) -> Child { let bin = workspace_target_bin("maestro-daemon", "maestrod"); - Command::new(&bin) - .env_clear() + let mut cmd = Command::new(&bin); + cmd.env_clear() .env( "PATH", std::env::var("PATH").unwrap_or_else(|_| "/usr/bin:/bin".into()), @@ -160,7 +208,35 @@ fn launch( .env("RUST_LOG", "warn") .stdout(Stdio::null()) .stderr(Stdio::inherit()) - .arg("run") - .spawn() + .arg("run"); + if let Some(p) = &opts.perimeter_id { + cmd.env("MAESTRO_DAEMON_PERIMETER_ID", p); + } + if let Some(a) = &opts.private_addr { + cmd.env("MAESTRO_DAEMON_PRIVATE_ADDR", a); + } + cmd.spawn() .unwrap_or_else(|e| panic!("spawn {} failed: {e}", bin.display())) } + +/// Write a minimal `~/.config/maestro/config.toml` carrying the NAP-247 +/// `[daemon.policy] allow_inbound` list. Everything else still comes from +/// env vars set by `launch`. +fn write_config(home: &std::path::Path, opts: &DaemonOptions) { + if opts.allow_inbound.is_empty() { + return; + } + let dir = home.join(".config/maestro"); + std::fs::create_dir_all(&dir).unwrap(); + let mut allow = String::new(); + for (i, id) in opts.allow_inbound.iter().enumerate() { + if i > 0 { + allow.push_str(", "); + } + allow.push('"'); + allow.push_str(id); + allow.push('"'); + } + let body = format!("[daemon.policy]\nallow_inbound = [{allow}]\n"); + std::fs::write(dir.join("config.toml"), body).unwrap(); +} diff --git a/crates/maestro-e2e/src/lib.rs b/crates/maestro-e2e/src/lib.rs index fa6d85a..eed3af1 100644 --- a/crates/maestro-e2e/src/lib.rs +++ b/crates/maestro-e2e/src/lib.rs @@ -16,5 +16,5 @@ pub mod util; pub use build::workspace_target_bin; pub use client::TestClient; -pub use daemon::Daemon; +pub use daemon::{Daemon, DaemonOptions}; pub use registry::Registry; diff --git a/crates/maestro-e2e/tests/e2e.rs b/crates/maestro-e2e/tests/e2e.rs index 0fd603d..984503b 100644 --- a/crates/maestro-e2e/tests/e2e.rs +++ b/crates/maestro-e2e/tests/e2e.rs @@ -15,7 +15,7 @@ use std::time::{Duration, Instant}; use futures_util::{SinkExt, StreamExt}; use maestro_e2e::util::wait_for_async; -use maestro_e2e::{Daemon, Registry, TestClient}; +use maestro_e2e::{Daemon, DaemonOptions, Registry, TestClient}; use maestro_protocol::{CreateSessionRequest, WorktreeSpec}; use tempfile::TempDir; use tokio_tungstenite::tungstenite::Message as WsMessage; @@ -401,3 +401,244 @@ fn git_worktree_list(dir: &std::path::Path) -> Vec { .map(str::to_string) .collect() } + +#[tokio::test(flavor = "multi_thread")] +async fn broker_t2_attach_through_home_daemon() { + // NAP-247: two daemons share perimeter "wg-test". alpha's broker resolves + // a session owned by beta, gets a pair JWT + perimeter address from the + // registry, dials beta via T2, and proxies the attach. + 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-test".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 { + perimeter_id: Some("wg-test".into()), + private_addr: Some(alpha_bind.to_string()), + allow_inbound: vec![], + }, + ) + .await; + assert_eq!(beta.bind_addr, beta_bind); + assert_eq!(alpha.bind_addr, alpha_bind); + + 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.iter().any(|d| d.daemon_id == "alpha") + && ds.iter().any(|d| d.daemon_id == "beta") + }) + .unwrap_or(false) + } + }) + .await; + + let cwd = TempDir::new().unwrap(); + let session = client + .spawn_session(&create_req("beta", cwd.path(), "bash --noprofile --norc")) + .await + .expect("spawn beta"); + + // Wait until alpha's catalog mirror sees the peer session — otherwise the + // broker can't locate the owning daemon. + let alpha_base = format!("http://{}", alpha.bind_addr); + let sid = session.id; + 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 && s.daemon_id == "beta") + } + }, + ) + .await; + + // Attach through alpha's broker. + let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid); + let (mut ws, _) = tokio_tungstenite::connect_async(broker_url) + .await + .expect("broker WS handshake"); + ws.send(WsMessage::Binary(b"echo MAESTRO_BROKER_E2E\n".to_vec())) + .await + .expect("send keys"); + + let want = b"MAESTRO_BROKER_E2E"; + let mut buf = Vec::new(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while std::time::Instant::now() < deadline { + tokio::select! { + msg = ws.next() => match msg { + Some(Ok(WsMessage::Binary(b))) => buf.extend(b), + Some(Ok(_)) => {} + _ => break, + }, + _ = tokio::time::sleep(Duration::from_millis(50)) => {} + } + if buf.windows(want.len()).any(|w| w == want) { + break; + } + } + assert!( + buf.windows(want.len()).any(|w| w == want), + "broker WS never echoed our keystrokes: buf len {}", + buf.len() + ); + + let _ = ws.close(None).await; + client.kill_session(sid).await.expect("kill"); + drop(alpha); + drop(beta); + drop(registry); +} + +#[tokio::test(flavor = "multi_thread")] +async fn broker_rejects_when_target_doesnt_allow_inbound() { + // NAP-247: beta has an empty allow_inbound, so beta's + // /sessions/:id/stream rejects alpha's broker JWT even though the + // registry happily mints one (default deny is target-side only). + 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-test".into()), + private_addr: Some(beta_bind.to_string()), + // Default deny — alpha not listed. + allow_inbound: vec![], + }, + ) + .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 { + perimeter_id: Some("wg-test".into()), + private_addr: Some(alpha_bind.to_string()), + allow_inbound: vec![], + }, + ) + .await; + assert_eq!(beta.bind_addr, beta_bind); + assert_eq!(alpha.bind_addr, alpha_bind); + + 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 alpha_base = format!("http://{}", alpha.bind_addr); + let sid = session.id; + 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; + + let broker_url = format!("ws://{}/api/attach/{}", alpha.bind_addr, sid); + // The handshake itself succeeds (alpha's broker accepts and starts + // proxying), but beta will reject the pair JWT — the upstream WS closes + // immediately, and alpha's proxy closes the client WS in turn. + let (mut ws, _) = tokio_tungstenite::connect_async(broker_url) + .await + .expect("alpha handshake"); + // We should get a close shortly without any binary frames. + let deadline = std::time::Instant::now() + Duration::from_secs(3); + let mut got_binary = false; + while std::time::Instant::now() < deadline { + tokio::select! { + msg = ws.next() => match msg { + Some(Ok(WsMessage::Binary(_))) => got_binary = true, + None | Some(Err(_)) => break, + Some(Ok(_)) => {} + }, + _ = tokio::time::sleep(Duration::from_millis(50)) => {} + } + } + assert!( + !got_binary, + "beta rejected the JWT but alpha somehow forwarded PTY bytes" + ); + + client.kill_session(sid).await.expect("kill"); + drop(alpha); + drop(beta); + drop(registry); +}