From cebb2d74b1b8cae3e2cce0e4c8cae0b86de6ce65 Mon Sep 17 00:00:00 2001 From: naps62-yolo Date: Tue, 26 May 2026 20:32:29 +0000 Subject: [PATCH] feat(maestro): daemon broker T2 + pair-JWT verifier + allow_inbound (NAP-247) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * New PairAttachVerifier (crates/maestro-daemon/src/jwt.rs): checks signature, aud == self, exp > now, and sub ∈ allow_inbound. Default deny — empty allow_inbound rejects every cross-daemon attach. * /sessions/:id/stream tries the legacy verifier first, falls back to the pair verifier on mismatch. Compat with existing direct-attach. * New BrokerHandle (crates/maestro-daemon/src/broker.rs): shared between the registry-client task and HTTP handlers; calls ResolveAttach over the control link and correlates the reply via a pending map. * RegistryClient now owns a BrokerHandle, drains its outbound mpsc alongside session events / heartbeats, and routes ResolveAttachResult back into the pending map. First heartbeat fires immediately so the registry has the daemon's perimeter info before the first ResolveAttach. * /api/attach/:id: self-owned → local PTY; peer-owned (per catalog mirror) → ResolveAttach + T2 dial + bidirectional frame proxy. Returns structured errors for missing local perimeter, peer on a different perimeter, etc. T1/T3 stay stubbed. Co-Authored-By: Claude Opus 4.7 (1M context) --- Cargo.lock | 7 + crates/maestro-daemon/Cargo.toml | 1 + crates/maestro-daemon/src/attach.rs | 58 ++++-- crates/maestro-daemon/src/broker.rs | 103 +++++++++++ crates/maestro-daemon/src/jwt.rs | 153 +++++++++++++++- crates/maestro-daemon/src/lib.rs | 4 +- crates/maestro-daemon/src/local_api.rs | 180 +++++++++++++++++-- crates/maestro-daemon/src/main.rs | 20 ++- crates/maestro-daemon/src/registry_client.rs | 45 ++++- crates/maestro-daemon/tests/attach.rs | 5 +- crates/maestro-daemon/tests/local_api.rs | 15 +- 11 files changed, 545 insertions(+), 46 deletions(-) create mode 100644 crates/maestro-daemon/src/broker.rs diff --git a/Cargo.lock b/Cargo.lock index d2da7e9..ff2a09a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -999,6 +999,7 @@ dependencies = [ "toml", "tracing", "tracing-subscriber", + "urlencoding", "uuid", ] @@ -2258,6 +2259,12 @@ dependencies = [ "serde", ] +[[package]] +name = "urlencoding" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" + [[package]] name = "utf-8" version = "0.7.6" diff --git a/crates/maestro-daemon/Cargo.toml b/crates/maestro-daemon/Cargo.toml index 0383b34..33f9aff 100644 --- a/crates/maestro-daemon/Cargo.toml +++ b/crates/maestro-daemon/Cargo.toml @@ -32,6 +32,7 @@ tokio-tungstenite = { workspace = true, features = ["rustls-tls-webpki-roots"] } toml.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +urlencoding = "2" uuid.workspace = true [dev-dependencies] diff --git a/crates/maestro-daemon/src/attach.rs b/crates/maestro-daemon/src/attach.rs index 91b1219..5b16a5c 100644 --- a/crates/maestro-daemon/src/attach.rs +++ b/crates/maestro-daemon/src/attach.rs @@ -18,13 +18,18 @@ use serde::Deserialize; use tokio::sync::broadcast::error::RecvError; use uuid::Uuid; -use crate::jwt::JwtVerifier; +use crate::jwt::{JwtVerifier, PairAttachVerifier}; use crate::session::SessionManager; #[derive(Debug, Clone)] pub struct AttachState { pub manager: SessionManager, pub verifier: Arc, + /// Pair-scoped JWT verifier (NAP-247). When the legacy verifier fails + /// (e.g. the token was minted as a pair-attach JWT for the broker path), + /// we fall back to this one before rejecting the client outright. Local + /// attaches always go through the legacy verifier. + pub pair_verifier: Arc, } #[derive(Debug, Deserialize)] @@ -32,12 +37,17 @@ struct AttachParams { token: String, } -pub fn router(manager: SessionManager, verifier: JwtVerifier) -> Router { +pub fn router( + manager: SessionManager, + verifier: JwtVerifier, + pair_verifier: PairAttachVerifier, +) -> Router { Router::new() .route("/sessions/:id/stream", any(attach_ws)) .with_state(AttachState { manager, verifier: Arc::new(verifier), + pair_verifier: Arc::new(pair_verifier), }) } @@ -47,18 +57,38 @@ async fn attach_ws( State(state): State, ws: WebSocketUpgrade, ) -> Result { - let claims = state - .verifier - .verify(¶ms.token) - .map_err(|e| (StatusCode::UNAUTHORIZED, e.to_string()))?; - if claims.session_id != id { - return Err(( - StatusCode::FORBIDDEN, - format!( - "token is for session {} but path requested {}", - claims.session_id, id - ), - )); + // Try the legacy `JwtClaims` shape first (browser/CLI direct attach). On + // any failure, try the pair-scoped shape used by the home-daemon broker + // (NAP-247). One — and only one — must accept. + let legacy_err = match state.verifier.verify(¶ms.token) { + Ok(claims) => { + if claims.session_id != id { + return Err(( + StatusCode::FORBIDDEN, + format!( + "token is for session {} but path requested {}", + claims.session_id, id + ), + )); + } + None + } + Err(e) => Some(e), + }; + if let Some(legacy_err) = legacy_err { + match state.pair_verifier.verify(¶ms.token, id) { + Ok(sub) => { + tracing::debug!(home_daemon = %sub, %id, "pair-scoped attach accepted"); + } + Err(pair_err) => { + // Both rejected — return the legacy error since most clients + // still mint that shape today. + return Err(( + StatusCode::UNAUTHORIZED, + format!("legacy: {legacy_err}; pair: {pair_err}"), + )); + } + } } if !state.manager.has(id).await { return Err((StatusCode::NOT_FOUND, format!("no session {id}"))); diff --git a/crates/maestro-daemon/src/broker.rs b/crates/maestro-daemon/src/broker.rs new file mode 100644 index 0000000..2cd880c --- /dev/null +++ b/crates/maestro-daemon/src/broker.rs @@ -0,0 +1,103 @@ +//! Home-daemon broker handle (NAP-247). +//! +//! The HTTP `WS /api/attach/:id` handler uses this to ask the registry to +//! resolve a peer-owned session into an [`AttachCandidates`] (a pair-scoped +//! JWT + the target daemon's T2 perimeter info). The actual control-link +//! send and the reply correlation live here so the handler stays thin. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use maestro_protocol::{AttachCandidates, DaemonOutbound, ReplyTo, RpcResult}; +use tokio::sync::{mpsc, oneshot, Mutex}; +use uuid::Uuid; + +/// How long to wait for a `ResolveAttachResult` after the request goes out. +/// Matches the registry's RPC timeout — a healthy registry replies in ms. +pub const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10); + +#[derive(Debug, thiserror::Error)] +pub enum BrokerError { + #[error("registry control link unavailable")] + NoControlLink, + #[error("registry reply channel closed before response")] + ReplyClosed, + #[error("registry did not reply within {0:?}")] + Timeout(Duration), + #[error("registry rejected resolve: {0}: {1}")] + Registry(String, String), +} + +/// Shared between the registry-client task (producer of `ResolveAttachResult`) +/// and the broker HTTP handler (consumer). Clone freely — all fields are +/// behind `Arc`s. +#[derive(Debug, Clone)] +pub struct BrokerHandle { + out_tx: mpsc::Sender, + pending: Arc>>>>, +} + +impl BrokerHandle { + /// `(handle, receiver)` — the registry-client task owns the receiver and + /// drains it into the control-link WS sink. The handle is shared with + /// HTTP handlers and is what they use to issue requests. + pub fn new() -> (Self, mpsc::Receiver) { + let (out_tx, out_rx) = mpsc::channel(64); + let handle = Self { + out_tx, + pending: Arc::new(Mutex::new(HashMap::new())), + }; + (handle, out_rx) + } + + /// Resolve a peer session: ask the registry for an [`AttachCandidates`]. + pub async fn resolve_attach(&self, session_id: Uuid) -> Result { + let reply_to = Uuid::new_v4(); + let (tx, rx) = oneshot::channel(); + self.pending.lock().await.insert(reply_to, tx); + + if self + .out_tx + .send(DaemonOutbound::ResolveAttach { + reply_to, + session_id, + }) + .await + .is_err() + { + self.pending.lock().await.remove(&reply_to); + return Err(BrokerError::NoControlLink); + } + + match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await { + Ok(Ok(RpcResult::Ok { value })) => Ok(value), + Ok(Ok(RpcResult::Err { error })) => { + Err(BrokerError::Registry(error.code, error.message)) + } + Ok(Err(_)) => Err(BrokerError::ReplyClosed), + Err(_) => { + self.pending.lock().await.remove(&reply_to); + Err(BrokerError::Timeout(RESOLVE_TIMEOUT)) + } + } + } + + /// Called by the registry-client task when a `ResolveAttachResult` frame + /// comes back from the registry. Resolves the waiting `resolve_attach` + /// future if one's still alive. + pub async fn complete_resolve( + &self, + reply_to: ReplyTo, + result: RpcResult, + ) -> bool { + let mut pending = self.pending.lock().await; + match pending.remove(&reply_to) { + Some(tx) => tx.send(result).is_ok(), + None => { + tracing::debug!(?reply_to, "no waiter for ResolveAttachResult"); + false + } + } + } +} diff --git a/crates/maestro-daemon/src/jwt.rs b/crates/maestro-daemon/src/jwt.rs index 3f662c1..3f4aacc 100644 --- a/crates/maestro-daemon/src/jwt.rs +++ b/crates/maestro-daemon/src/jwt.rs @@ -2,8 +2,11 @@ //! validated locally by each daemon using the shared secret — no callback to //! the registry on the attach hot path. Spec §Auth. +use std::collections::HashSet; + use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation}; -use maestro_protocol::JwtClaims; +use maestro_protocol::{JwtClaims, PairAttachClaims}; +use uuid::Uuid; #[derive(Debug, thiserror::Error)] pub enum JwtError { @@ -11,6 +14,10 @@ pub enum JwtError { Decode(#[from] jsonwebtoken::errors::Error), #[error("daemon_id mismatch (token wants `{wanted}`, daemon is `{ours}`)")] WrongDaemon { wanted: String, ours: String }, + #[error("session mismatch (token wants `{wanted}`, path is `{path}`)")] + WrongSession { wanted: Uuid, path: Uuid }, + #[error("home daemon `{sub}` not in this daemon's allow_inbound")] + NotAllowed { sub: String }, } #[derive(Clone)] @@ -59,6 +66,67 @@ pub fn issue(secret: &str, claims: &JwtClaims) -> Result, +} + +impl std::fmt::Debug for PairAttachVerifier { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("PairAttachVerifier") + .field("daemon_id", &self.daemon_id) + .field("allow_inbound", &self.allow_inbound) + .field("decoding", &"") + .finish() + } +} + +impl PairAttachVerifier { + pub fn new( + secret: &str, + daemon_id: impl Into, + allow_inbound: impl IntoIterator, + ) -> Self { + Self { + decoding: DecodingKey::from_secret(secret.as_bytes()), + daemon_id: daemon_id.into(), + allow_inbound: allow_inbound.into_iter().collect(), + } + } + + /// Validate the token for the path-bound `session_id`. On success returns + /// the home daemon's id (`sub`) for logging/tracing. + pub fn verify(&self, token: &str, session_id: Uuid) -> Result { + let mut validation = Validation::new(Algorithm::HS256); + // Require explicit audience match — the jsonwebtoken crate validates + // `aud` against this set on decode. + validation.set_audience(&[self.daemon_id.as_str()]); + let data = decode::(token, &self.decoding, &validation)?; + let claims = data.claims; + if claims.aud != self.daemon_id { + return Err(JwtError::WrongDaemon { + wanted: claims.aud, + ours: self.daemon_id.clone(), + }); + } + if claims.sid != session_id { + return Err(JwtError::WrongSession { + wanted: claims.sid, + path: session_id, + }); + } + if !self.allow_inbound.contains(&claims.sub) { + return Err(JwtError::NotAllowed { sub: claims.sub }); + } + Ok(claims.sub) + } +} + #[cfg(test)] mod tests { use super::*; @@ -110,4 +178,87 @@ mod tests { let token = issue("wrong", &claims_for("laptop", 300)).unwrap(); assert!(v.verify(&token).is_err()); } + + // ----- PairAttachVerifier ----- + + fn pair_claims(home: &str, target: &str, sid: Uuid, valid_seconds: i64) -> PairAttachClaims { + PairAttachClaims { + iss: "registry".into(), + sub: home.into(), + aud: target.into(), + sid, + exp: (Utc::now() + Duration::seconds(valid_seconds)).timestamp(), + } + } + + fn issue_pair(secret: &str, claims: &PairAttachClaims) -> String { + let header = Header::new(Algorithm::HS256); + let key = EncodingKey::from_secret(secret.as_bytes()); + encode(&header, claims, &key).unwrap() + } + + #[test] + fn pair_verifier_accepts_allow_listed_home() { + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]); + let sid = Uuid::new_v4(); + let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60)); + let sub = v.verify(&token, sid).unwrap(); + assert_eq!(sub, "laptop"); + } + + #[test] + fn pair_verifier_rejects_unlisted_home() { + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", ["other".into()]); + let sid = Uuid::new_v4(); + let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60)); + let err = v.verify(&token, sid).unwrap_err(); + assert!(matches!(err, JwtError::NotAllowed { .. }), "{err:?}"); + } + + #[test] + fn pair_verifier_rejects_empty_allow_inbound() { + // Default deny: no peer listed → every cross-daemon attach fails. + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", std::iter::empty::()); + let sid = Uuid::new_v4(); + let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60)); + assert!(matches!( + v.verify(&token, sid).unwrap_err(), + JwtError::NotAllowed { .. } + )); + } + + #[test] + fn pair_verifier_rejects_wrong_audience() { + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]); + let sid = Uuid::new_v4(); + // Token addressed to a different target — must not pass our checks. + let token = issue_pair(secret, &pair_claims("laptop", "elsewhere", sid, 60)); + assert!(v.verify(&token, sid).is_err()); + } + + #[test] + fn pair_verifier_rejects_session_id_mismatch() { + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]); + let sid = Uuid::new_v4(); + let path = Uuid::new_v4(); + let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, 60)); + assert!(matches!( + v.verify(&token, path).unwrap_err(), + JwtError::WrongSession { .. } + )); + } + + #[test] + fn pair_verifier_rejects_expired() { + let secret = "s"; + let v = PairAttachVerifier::new(secret, "konishi", ["laptop".into()]); + let sid = Uuid::new_v4(); + let token = issue_pair(secret, &pair_claims("laptop", "konishi", sid, -120)); + assert!(v.verify(&token, sid).is_err()); + } } diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 048ec4f..319c48e 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -7,6 +7,7 @@ #![deny(missing_debug_implementations)] pub mod attach; +pub mod broker; pub mod catalog_mirror; pub mod config; pub mod configure; @@ -19,9 +20,10 @@ pub mod sidecar; pub mod tmux; pub mod worktree; +pub use broker::{BrokerError, BrokerHandle}; pub use catalog_mirror::CatalogMirror; pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides}; pub use configure::{run as run_configure, ConfigureError, ConfigureInputs}; -pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier}; +pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier, PairAttachVerifier}; pub use registry_client::{ClientError, RegistryClient, HEARTBEAT_INTERVAL}; pub use session::{SessionError, SessionEvent, SessionManager}; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index fe1c10b..df0a6e9 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -179,33 +179,185 @@ pub fn state_for( // Broker WS — `WS /api/attach/:session_id` // --------------------------------------------------------------------------- // -// v1 only handles the self case: if this daemon owns the session, delegate to -// the existing PTY subscribe path. For peer-owned sessions we reply with a -// structured 501 — NAP-247 (T2 broker) implements the cross-daemon path. +// Home-daemon broker (NAP-243 + NAP-247): +// +// 1. Session owned by self → delegate to local PTY subscribe. +// 2. Session owned by a peer daemon → ResolveAttach via the registry, dial +// the peer over T2, proxy frames. +// 3. Unknown session → 404. -pub fn broker_router(manager: SessionManager) -> Router { +use crate::broker::{BrokerError, BrokerHandle}; + +#[derive(Debug, Clone)] +pub struct BrokerState { + pub manager: SessionManager, + pub catalog: CatalogMirror, + pub broker: BrokerHandle, + pub self_daemon_id: Arc, + pub self_perimeter_id: Option>, +} + +pub fn broker_router(state: BrokerState) -> Router { Router::new() .route("/api/attach/:id", any(broker_attach_ws)) - .with_state(manager) + .with_state(state) } async fn broker_attach_ws( Path(id): Path, - State(manager): State, + State(state): State, ws: axum::extract::WebSocketUpgrade, ) -> Result { - if !manager.has(id).await { - // Peer-owned sessions land here once NAP-246 (peer catalog mirror) - // ships. Until then we can't even tell them apart from "doesn't exist" - // — both produce a 404. NAP-247 will distinguish the two. + if state.manager.has(id).await { + // Local short-circuit. + let manager = state.manager.clone(); + return Ok(ws.on_upgrade(move |socket| async move { + let _ = crate::attach::serve_local(socket, manager, id).await; + })); + } + + // Look up which peer owns the session via the catalog mirror. + let peer = state + .catalog + .sessions() + .await + .into_iter() + .find(|s| s.id == id) + .map(|s| s.daemon_id); + let Some(peer_daemon_id) = peer else { return Err(error_response( StatusCode::NOT_FOUND, "unknown_session", - "no such session on this daemon (peer broker is NAP-247)", + "no such session on this daemon or any known peer", )); - } - let upgrade_manager = manager.clone(); + }; + + // ResolveAttach via the control link. + let candidates = match state.broker.resolve_attach(id).await { + Ok(c) => c, + Err(e) => { + tracing::warn!(%id, %peer_daemon_id, error = %e, "ResolveAttach failed"); + let (code, msg) = broker_error_code(&e); + return Err(error_response(StatusCode::BAD_GATEWAY, code, &msg)); + } + }; + + // Pick a transport. T1 is canceled (NAP-235); T3 is gated (NAP-238). + // For T2 we need both daemons to declare the same perimeter id, and the + // candidate must include a private address. + let perimeter = candidates.perimeter.clone(); + let our_perimeter = state.self_perimeter_id.as_deref().map(|s| s.as_str()); + let dial_url = match (perimeter, our_perimeter) { + (Some(p), Some(ours)) if p.perimeter_id == ours => { + format!( + "ws://{}/sessions/{}/stream?token={}", + p.private_addr, + id, + urlencoding::encode(&candidates.jwt), + ) + } + (Some(_), Some(_)) => { + return Err(error_response( + StatusCode::BAD_GATEWAY, + "perimeter_mismatch", + "peer is on a different perimeter; T2 unavailable, T1/T3 not implemented", + )); + } + (Some(_), None) => { + return Err(error_response( + StatusCode::BAD_GATEWAY, + "no_local_perimeter", + "this daemon has no perimeter_id set; can't use T2", + )); + } + (None, _) => { + return Err(error_response( + StatusCode::BAD_GATEWAY, + "no_perimeter_candidate", + "peer hasn't advertised a perimeter; T2 unavailable", + )); + } + }; + Ok(ws.on_upgrade(move |socket| async move { - let _ = crate::attach::serve_local(socket, upgrade_manager, id).await; + let _ = pump_broker_ws(socket, dial_url).await; })) } + +/// Bidirectional pump between the client WS and the peer daemon's WS. +async fn pump_broker_ws( + client: axum::extract::ws::WebSocket, + upstream_url: String, +) -> Result<(), axum::Error> { + use axum::extract::ws::Message as AxMessage; + use futures_util::{SinkExt, StreamExt}; + use tokio_tungstenite::tungstenite::Message as WsMessage; + + let (upstream, _) = match tokio_tungstenite::connect_async(&upstream_url).await { + Ok(pair) => pair, + Err(e) => { + tracing::warn!(url = %upstream_url, error = %e, "T2 dial failed"); + return Ok(()); + } + }; + + let (mut client_sink, mut client_stream) = client.split(); + let (mut up_sink, mut up_stream) = upstream.split(); + + let client_to_up = async { + while let Some(Ok(msg)) = client_stream.next().await { + let frame = match msg { + AxMessage::Text(t) => WsMessage::Text(t), + AxMessage::Binary(b) => WsMessage::Binary(b), + AxMessage::Ping(p) => WsMessage::Ping(p), + AxMessage::Pong(p) => WsMessage::Pong(p), + AxMessage::Close(_) => break, + }; + if up_sink.send(frame).await.is_err() { + break; + } + } + }; + let up_to_client = async { + while let Some(Ok(msg)) = up_stream.next().await { + let frame = match msg { + WsMessage::Text(t) => AxMessage::Text(t), + WsMessage::Binary(b) => AxMessage::Binary(b), + WsMessage::Ping(p) => AxMessage::Ping(p), + WsMessage::Pong(p) => AxMessage::Pong(p), + WsMessage::Close(_) | WsMessage::Frame(_) => break, + }; + if client_sink.send(frame).await.is_err() { + break; + } + } + }; + + tokio::select! { + _ = client_to_up => {} + _ = up_to_client => {} + } + Ok(()) +} + +fn broker_error_code(e: &BrokerError) -> (&'static str, String) { + match e { + BrokerError::NoControlLink => ( + "no_registry_link", + "this daemon is not connected to the registry".into(), + ), + BrokerError::ReplyClosed => ("reply_closed", e.to_string()), + BrokerError::Timeout(_) => ("resolve_timeout", e.to_string()), + BrokerError::Registry(code, msg) => ( + // Code points are stable across the protocol so the client can + // distinguish "unknown_session" from "self_target" etc. + match code.as_str() { + "unknown_session" => "unknown_session", + "self_target" => "self_target", + "jwt_failed" => "jwt_failed", + _ => "registry_error", + }, + msg.clone(), + ), + } +} diff --git a/crates/maestro-daemon/src/main.rs b/crates/maestro-daemon/src/main.rs index 8f21c51..af5bcf4 100644 --- a/crates/maestro-daemon/src/main.rs +++ b/crates/maestro-daemon/src/main.rs @@ -15,7 +15,7 @@ use maestro_daemon::tmux::Tmux; use maestro_daemon::worktree::WorktreeManager; use maestro_daemon::{ attach, default_config_path, default_state_dir, load_config, local_api, run_configure, - ConfigureInputs, EnvOverrides, JwtVerifier, RegistryClient, SessionManager, + ConfigureInputs, EnvOverrides, JwtVerifier, PairAttachVerifier, RegistryClient, SessionManager, }; use tokio::net::TcpListener; use tracing_subscriber::EnvFilter; @@ -133,6 +133,11 @@ async fn run(config_override: Option) -> Result<(), Box> { manager.start_liveness_watcher(DEFAULT_LIVENESS_INTERVAL); let verifier = JwtVerifier::new(&config.token, &config.daemon_id); + let pair_verifier = PairAttachVerifier::new( + &config.token, + &config.daemon_id, + config.allow_inbound.iter().cloned(), + ); // Build the registry client first so we can hand its catalog mirror to // the local-API state — both halves share the same Arc>. @@ -151,11 +156,18 @@ async fn run(config_override: Option) -> Result<(), Box> { manager.clone(), config.hostname.clone(), config.daemon_url.clone(), - catalog, + catalog.clone(), ); - let router = attach::router(manager.clone(), verifier) + let broker_state = local_api::BrokerState { + manager: manager.clone(), + catalog, + broker: client.broker(), + self_daemon_id: std::sync::Arc::new(config.daemon_id.clone()), + self_perimeter_id: config.perimeter_id.clone().map(std::sync::Arc::new), + }; + let router = attach::router(manager.clone(), verifier, pair_verifier) .merge(local_api::router(local_state)) - .merge(local_api::broker_router(manager.clone())); + .merge(local_api::broker_router(broker_state)); let listener = TcpListener::bind(config.bind_addr).await?; let bound = listener.local_addr()?; diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index 4f8f7a4..c9ea366 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -17,6 +17,9 @@ use maestro_protocol::{DaemonInbound, DaemonOutbound, RpcResult, PROTOCOL_VERSIO use tokio::sync::broadcast::error::RecvError; use tokio_tungstenite::tungstenite::Message; +use tokio::sync::mpsc; + +use crate::broker::BrokerHandle; use crate::catalog_mirror::CatalogMirror; use crate::session::{SessionEvent, SessionManager}; @@ -37,7 +40,7 @@ pub enum ClientError { Handshake(String), } -#[derive(Debug, Clone)] +#[derive(Debug)] pub struct RegistryClient { registry_url: String, daemon_id: String, @@ -54,6 +57,13 @@ pub struct RegistryClient { /// populated from `DaemonInbound::CatalogEvent` frames the registry /// fans out. catalog: CatalogMirror, + /// Shared with HTTP handlers (NAP-247): when one calls `resolve_attach`, + /// we route the request through this client's control link and route the + /// reply back via the handle's pending map. + broker: BrokerHandle, + /// Outbound channel from `BrokerHandle::resolve_attach` (and future broker + /// requests). Drained inside `connect_once` and forwarded into the WS. + broker_rx: mpsc::Receiver, } impl RegistryClient { @@ -65,6 +75,7 @@ impl RegistryClient { token: impl Into, manager: SessionManager, ) -> Self { + let (broker, broker_rx) = BrokerHandle::new(); Self { registry_url: registry_url.into(), daemon_id: daemon_id.into(), @@ -75,9 +86,17 @@ impl RegistryClient { private_addr: None, manager, catalog: CatalogMirror::new(), + broker, + broker_rx, } } + /// Cheap clone of the broker handle for HTTP handlers (NAP-247). Sends + /// here are pumped through this client's control link. + pub fn broker(&self) -> BrokerHandle { + self.broker.clone() + } + /// Cheap handle to the daemon's peer-catalog mirror. Clone freely; /// internal state is `Arc>`. pub fn catalog(&self) -> CatalogMirror { @@ -98,7 +117,7 @@ impl RegistryClient { /// Run forever, reconnecting with exponential backoff between attempts. /// Returns only if cancelled by the caller dropping the task. - pub async fn run(self) { + pub async fn run(mut self) { let mut backoff = BACKOFF_MIN; loop { match self.connect_once().await { @@ -115,7 +134,7 @@ impl RegistryClient { } } - async fn connect_once(&self) -> Result<(), ClientError> { + async fn connect_once(&mut self) -> Result<(), ClientError> { let (ws, _) = tokio_tungstenite::connect_async(&self.registry_url).await?; let (mut sink, mut stream) = ws.split(); @@ -157,9 +176,11 @@ impl RegistryClient { .await?; let mut events = self.manager.events(); + // The first interval tick fires immediately, so the registry gets a + // perimeter advertisement (NAP-247) right after the handshake instead + // of waiting a full HEARTBEAT_INTERVAL for `update_perimeter` to land. let mut hb = tokio::time::interval(HEARTBEAT_INTERVAL); hb.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - let _ = hb.tick().await; // consume the immediate first tick loop { tokio::select! { @@ -203,6 +224,11 @@ impl RegistryClient { }, )?)).await?; } + Some(broker_msg) = self.broker_rx.recv() => { + // Outbound broker requests (NAP-247) — e.g. ResolveAttach. + sink.send(Message::Text(serde_json::to_string(&broker_msg)?)) + .await?; + } } } Ok(()) @@ -242,10 +268,13 @@ impl RegistryClient { )?)) .await?; } - DaemonInbound::ResolveAttachResult { .. } | DaemonInbound::RelaySetup { .. } => { - // Wired by NAP-243 (broker mode). Lands here only so registries - // emitting these messages don't crash older daemons on parse. - tracing::debug!("broker message received but unhandled (NAP-243 pending)"); + DaemonInbound::ResolveAttachResult { reply_to, result } => { + self.broker.complete_resolve(reply_to, result).await; + } + DaemonInbound::RelaySetup { .. } => { + // T3 (relay) is gated on NAP-238. Lands here only so newer + // registries sending it don't crash this daemon on parse. + tracing::debug!("RelaySetup received but T3 not implemented (NAP-238 pending)"); } DaemonInbound::CatalogEvent { event } => { // NAP-246: peer-filtered catalog delta from the registry. diff --git a/crates/maestro-daemon/tests/attach.rs b/crates/maestro-daemon/tests/attach.rs index 0079083..3571665 100644 --- a/crates/maestro-daemon/tests/attach.rs +++ b/crates/maestro-daemon/tests/attach.rs @@ -12,7 +12,7 @@ use std::time::Duration; use chrono::{Duration as ChronoDuration, Utc}; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::{attach, issue_jwt, JwtVerifier, SessionManager}; +use maestro_daemon::{attach, issue_jwt, JwtVerifier, PairAttachVerifier, SessionManager}; use maestro_protocol::{CreateSessionRequest, JwtClaims}; use tempfile::TempDir; use tokio::net::TcpListener; @@ -36,7 +36,8 @@ async fn spawn_daemon_ws(daemon_id: &str) -> Harness { let fx = DaemonFixture::new(daemon_id); let manager = fx.manager.clone(); let verifier = JwtVerifier::new(TOKEN, daemon_id); - let router = attach::router(manager.clone(), verifier); + let pair_verifier = PairAttachVerifier::new(TOKEN, daemon_id, std::iter::empty::()); + let router = attach::router(manager.clone(), verifier, pair_verifier); let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) .await .unwrap(); diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 90ef17d..8e3167e 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -11,7 +11,7 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::{local_api, CatalogMirror}; +use maestro_daemon::{local_api, BrokerHandle, CatalogMirror}; use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session}; use tokio::net::TcpListener; use tokio::sync::oneshot; @@ -34,7 +34,18 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { let manager = fx.manager.clone(); let catalog = CatalogMirror::new(); let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone()); - let router = local_api::router(state).merge(local_api::broker_router(manager.clone())); + // NAP-247: tests don't exercise the cross-host broker path; hand the + // router an isolated BrokerHandle whose receiver is dropped, so any + // resolve_attach attempt fails fast with NoControlLink. + let (broker, _broker_rx) = BrokerHandle::new(); + let broker_state = local_api::BrokerState { + manager: manager.clone(), + catalog: catalog.clone(), + broker, + self_daemon_id: std::sync::Arc::new(daemon_id.into()), + self_perimeter_id: None, + }; + let router = local_api::router(state).merge(local_api::broker_router(broker_state)); let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) .await .unwrap();