test(maestro): e2e for broker T2 + allow_inbound enforcement (NAP-247)

Two scenarios:

* broker_t2_attach_through_home_daemon: alpha + beta share perimeter
  "wg-test"; beta lists alpha in allow_inbound. Test attaches to beta's
  session via alpha's /api/attach/:id and verifies the PTY echoes back.
* broker_rejects_when_target_doesnt_allow_inbound: same setup but beta's
  allow_inbound is empty; the broker WS handshake succeeds at alpha
  (which has no reason to refuse) but beta drops the pair JWT, so the
  proxied stream closes without any PTY bytes flowing.

Extends maestro-e2e::Daemon with DaemonOptions and a spawn_at variant
that fixes the bind address, so the daemon's advertised private_addr
matches the port the broker will dial.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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<String>,
+    pub private_addr: Option<String>,
+    pub allow_inbound: Vec<String>,
+}
+
 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();
+}

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;

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<String> {
         .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",
+        &registry.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",
+        &registry.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(&registry.http_base(), &registry.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<maestro_protocol::Session> = 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",
+        &registry.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",
+        &registry.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(&registry.http_base(), &registry.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<maestro_protocol::Session> = 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);
+}