diff --git a/Cargo.lock b/Cargo.lock index d1ac0f4..0665ca8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -950,6 +950,7 @@ dependencies = [ "maestro-config", "maestro-protocol", "maestro-registry", + "reqwest", "serde", "serde_json", "tempfile", diff --git a/crates/maestro-daemon/Cargo.toml b/crates/maestro-daemon/Cargo.toml index 27833f9..0383b34 100644 --- a/crates/maestro-daemon/Cargo.toml +++ b/crates/maestro-daemon/Cargo.toml @@ -35,5 +35,6 @@ tracing-subscriber.workspace = true uuid.workspace = true [dev-dependencies] +reqwest.workspace = true tempfile.workspace = true maestro-registry = { path = "../maestro-registry" } diff --git a/crates/maestro-daemon/src/attach.rs b/crates/maestro-daemon/src/attach.rs index 63b7750..91b1219 100644 --- a/crates/maestro-daemon/src/attach.rs +++ b/crates/maestro-daemon/src/attach.rs @@ -65,11 +65,14 @@ async fn attach_ws( } let manager = state.manager.clone(); Ok(ws.on_upgrade(move |socket| async move { - let _ = serve_attach(socket, manager, id).await; + let _ = serve_local(socket, manager, id).await; })) } -async fn serve_attach( +/// Pump a WebSocket against a local session's PTY. Reused by both the legacy +/// `/sessions/:id/stream` endpoint (after JWT auth) and the new broker +/// `WS /api/attach/:id` endpoint (when the session is local-owned — NAP-243). +pub(crate) async fn serve_local( socket: WebSocket, manager: SessionManager, id: Uuid, diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index e5463c7..d8d3a04 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -10,6 +10,7 @@ pub mod attach; pub mod config; pub mod configure; pub mod jwt; +pub mod local_api; pub mod pty; pub mod registry_client; pub mod session; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs new file mode 100644 index 0000000..102d79d --- /dev/null +++ b/crates/maestro-daemon/src/local_api.rs @@ -0,0 +1,201 @@ +//! Daemon-side HTTP API for local clients (NAP-243). +//! +//! These routes live on the daemon's existing `bind_addr` (default +//! `127.0.0.1:7070`) alongside the legacy `/sessions/:id/stream` WS endpoint. +//! They're the contract `maestro-web` (NAP-241) and the rewired CLI (NAP-244) +//! call against — a same-origin, same-trust surface that doesn't need a +//! registry round-trip for catalog reads. +//! +//! ## Trust model (v1) +//! +//! Whoever can reach `bind_addr` can use these routes. No JWT, no cookie. The +//! default `127.0.0.1` bind makes this same-trust as tmux (one-user-wide). +//! Operators who bind a perimeter address (LAN/WG IP) opt into one-segment-wide +//! trust — same model `maestro-web` documents per the connection-orchestration +//! spec. Cookie auth can be added later if a real use case demands it. +//! +//! ## Scope (v1) +//! +//! The daemon only surfaces its own sessions and itself. Peer-catalog mirroring +//! is NAP-246; cross-daemon broker attach is NAP-247. Until those land, +//! `POST /api/sessions` rejects requests targeting other daemons and +//! `WS /api/attach/:id` only handles sessions this daemon owns. + +use std::sync::Arc; + +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::{any, delete, get}; +use axum::{Json, Router}; +use maestro_protocol::{CreateSessionRequest, DaemonInfo, ErrorInfo, Session, PROTOCOL_VERSION}; +use serde::Serialize; +use uuid::Uuid; + +use crate::session::SessionManager; + +/// Daemon identity returned by `GET /api/web/info`. Browsers render this in the +/// header ("you're connected to konishi"); the field set mirrors `DaemonInfo` +/// plus the running daemon's protocol version. +#[derive(Debug, Clone, Serialize)] +pub struct WebInfo { + pub daemon_id: String, + pub hostname: String, + pub version: String, + pub daemon_url: String, +} + +#[derive(Debug, Clone)] +pub struct LocalApiState { + pub manager: SessionManager, + pub hostname: Arc, + pub daemon_url: Arc, +} + +pub fn router(state: LocalApiState) -> Router { + Router::new() + .route("/api/sessions", get(list_sessions).post(create_session)) + .route("/api/sessions/:id", delete(kill_session)) + .route("/api/daemons", get(list_daemons)) + .route("/api/web/info", get(web_info)) + .with_state(state) +} + +async fn list_sessions(State(s): State) -> Json> { + Json(s.manager.snapshot().await) +} + +async fn list_daemons(State(s): State) -> Json> { + // v1: a daemon only knows about itself. NAP-246 populates peers. + Json(vec![DaemonInfo { + daemon_id: s.manager.daemon_id().to_string(), + hostname: (*s.hostname).clone(), + version: PROTOCOL_VERSION.into(), + daemon_url: (*s.daemon_url).clone(), + }]) +} + +async fn web_info(State(s): State) -> Json { + Json(WebInfo { + daemon_id: s.manager.daemon_id().to_string(), + hostname: (*s.hostname).clone(), + version: PROTOCOL_VERSION.into(), + daemon_url: (*s.daemon_url).clone(), + }) +} + +async fn create_session( + State(s): State, + Json(body): Json, +) -> Response { + if body.daemon_id != s.manager.daemon_id() { + // Forwarding to peer daemons is NAP-247 (broker mode). Until then, + // tell the caller exactly why we refused — the registry's API still + // accepts cross-daemon creates the old way, so this isn't a regression. + return error_response( + StatusCode::NOT_IMPLEMENTED, + "remote_create_unsupported", + &format!( + "this daemon ({}) can only create sessions for itself; \ + cross-daemon broker is NAP-247", + s.manager.daemon_id() + ), + ); + } + match s.manager.create(body).await { + Ok(session) => (StatusCode::CREATED, Json(session)).into_response(), + Err(e) => { + let info = e.to_error_info(); + error_response( + StatusCode::BAD_GATEWAY, + info.code.as_str(), + info.message.as_str(), + ) + } + } +} + +async fn kill_session(State(s): State, Path(id): Path) -> Response { + if !s.manager.has(id).await { + // The session might exist on a peer daemon, but v1 doesn't broker — + // give a clean 404 rather than lying about success. + return error_response( + StatusCode::NOT_FOUND, + "unknown_session", + "no such session on this daemon", + ); + } + match s.manager.kill(id).await { + Ok(()) => StatusCode::NO_CONTENT.into_response(), + Err(e) => { + let info = e.to_error_info(); + error_response( + StatusCode::BAD_GATEWAY, + info.code.as_str(), + info.message.as_str(), + ) + } + } +} + +fn error_response(status: StatusCode, code: &str, message: &str) -> Response { + ( + status, + Json(serde_json::json!({ + "error": ErrorInfo { + code: code.into(), + message: message.into(), + }, + })), + ) + .into_response() +} + +/// Convenience constructor — pulls the bits `LocalApiState` needs from +/// `DaemonConfig` so callers don't have to assemble it by hand. +pub fn state_for( + manager: SessionManager, + hostname: impl Into, + daemon_url: impl Into, +) -> LocalApiState { + LocalApiState { + manager, + hostname: Arc::new(hostname.into()), + daemon_url: Arc::new(daemon_url.into()), + } +} + +// --------------------------------------------------------------------------- +// 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. + +pub fn broker_router(manager: SessionManager) -> Router { + Router::new() + .route("/api/attach/:id", any(broker_attach_ws)) + .with_state(manager) +} + +async fn broker_attach_ws( + Path(id): Path, + State(manager): 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. + return Err(error_response( + StatusCode::NOT_FOUND, + "unknown_session", + "no such session on this daemon (peer broker is NAP-247)", + )); + } + let upgrade_manager = manager.clone(); + Ok(ws.on_upgrade(move |socket| async move { + let _ = crate::attach::serve_local(socket, upgrade_manager, id).await; + })) +} diff --git a/crates/maestro-daemon/src/main.rs b/crates/maestro-daemon/src/main.rs index 25d07d7..705ea26 100644 --- a/crates/maestro-daemon/src/main.rs +++ b/crates/maestro-daemon/src/main.rs @@ -14,8 +14,8 @@ use maestro_daemon::sidecar::SidecarStore; use maestro_daemon::tmux::Tmux; use maestro_daemon::worktree::WorktreeManager; use maestro_daemon::{ - attach, default_config_path, default_state_dir, load_config, run_configure, ConfigureInputs, - EnvOverrides, JwtVerifier, RegistryClient, SessionManager, + attach, default_config_path, default_state_dir, load_config, local_api, run_configure, + ConfigureInputs, EnvOverrides, JwtVerifier, RegistryClient, SessionManager, }; use tokio::net::TcpListener; use tracing_subscriber::EnvFilter; @@ -133,7 +133,14 @@ 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 router = attach::router(manager.clone(), verifier); + let local_state = local_api::state_for( + manager.clone(), + config.hostname.clone(), + config.daemon_url.clone(), + ); + let router = attach::router(manager.clone(), verifier) + .merge(local_api::router(local_state)) + .merge(local_api::broker_router(manager.clone())); let listener = TcpListener::bind(config.bind_addr).await?; let bound = listener.local_addr()?; diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs new file mode 100644 index 0000000..8c0c22c --- /dev/null +++ b/crates/maestro-daemon/tests/local_api.rs @@ -0,0 +1,243 @@ +//! Integration tests for the daemon's local HTTP API + broker WS (NAP-243). +//! +//! The legacy `/sessions/:id/stream` endpoint is covered by `tests/attach.rs`; +//! this file exercises the new `/api/*` surface and the same-host broker WS. + +mod common; + +use std::collections::HashMap; +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; + +use common::DaemonFixture; +use futures_util::{SinkExt, StreamExt}; +use maestro_daemon::local_api; +use maestro_protocol::{CreateSessionRequest, DaemonInfo, Session}; +use tokio::net::TcpListener; +use tokio::sync::oneshot; +use tokio_tungstenite::tungstenite::Message as WsMessage; +use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use uuid::Uuid; + +type WsClient = WebSocketStream>; + +struct Harness { + addr: SocketAddr, + manager: maestro_daemon::SessionManager, + _shutdown: oneshot::Sender<()>, + _fx: DaemonFixture, +} + +async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness { + let fx = DaemonFixture::new(daemon_id); + let manager = fx.manager.clone(); + let state = local_api::state_for(manager.clone(), hostname, daemon_url); + let router = local_api::router(state).merge(local_api::broker_router(manager.clone())); + let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let (tx, rx) = oneshot::channel::<()>(); + tokio::spawn(async move { + let _ = axum::serve(listener, router) + .with_graceful_shutdown(async move { + let _ = rx.await; + }) + .await; + }); + Harness { + addr, + manager, + _shutdown: tx, + _fx: fx, + } +} + +fn url(addr: SocketAddr, path: &str) -> String { + format!("http://{addr}{path}") +} + +#[tokio::test] +async fn web_info_returns_daemon_identity() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let resp: serde_json::Value = reqwest::get(url(h.addr, "/api/web/info")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(resp["daemon_id"], "laptop"); + assert_eq!(resp["hostname"], "naps-laptop"); + assert_eq!(resp["daemon_url"], "ws://10.0.0.1:7070"); + assert!(resp["version"].is_string()); +} + +#[tokio::test] +async fn list_daemons_returns_self_only() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let list: Vec = reqwest::get(url(h.addr, "/api/daemons")) + .await + .unwrap() + .json() + .await + .unwrap(); + // v1: no peer mirror, so a daemon only knows about itself. + assert_eq!(list.len(), 1); + assert_eq!(list[0].daemon_id, "laptop"); +} + +#[tokio::test] +async fn list_sessions_starts_empty_and_reflects_creates() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + + let empty: Vec = reqwest::get(url(h.addr, "/api/sessions")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(empty.is_empty()); + + let req = CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: h.manager.daemon_id().into(), + command: "sleep 60".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + let created: Session = reqwest::Client::new() + .post(url(h.addr, "/api/sessions")) + .json(&req) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(created.daemon_id, "laptop"); + + let after: Vec = reqwest::get(url(h.addr, "/api/sessions")) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(after.len(), 1); + assert_eq!(after[0].id, created.id); +} + +#[tokio::test] +async fn post_to_peer_daemon_id_returns_501_not_proxy() { + // Cross-daemon broker is NAP-247; until it lands a daemon must refuse + // (rather than silently misroute) a create targeting a peer. + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let req = CreateSessionRequest { + daemon_id: "konishi".into(), + cwd: "/tmp".into(), + command: "sleep 60".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + let resp = reqwest::Client::new() + .post(url(h.addr, "/api/sessions")) + .json(&req) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::NOT_IMPLEMENTED); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["error"]["code"], "remote_create_unsupported"); +} + +#[tokio::test] +async fn delete_unknown_session_returns_404() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let resp = reqwest::Client::new() + .delete(url(h.addr, &format!("/api/sessions/{}", Uuid::new_v4()))) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn delete_owned_session_kills_it() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let session = h + .manager + .create(CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: "/tmp".into(), + command: "sleep 60".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .unwrap(); + let resp = reqwest::Client::new() + .delete(url(h.addr, &format!("/api/sessions/{}", session.id))) + .send() + .await + .unwrap(); + assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT); + assert!(!h.manager.has(session.id).await); +} + +#[tokio::test] +async fn broker_ws_attaches_to_local_session() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let session = h + .manager + .create(CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: "/tmp".into(), + command: "bash --noprofile --norc".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .unwrap(); + + let ws_url = format!("ws://{}/api/attach/{}", h.addr, session.id); + let (mut ws, _): (WsClient, _) = connect_async(ws_url).await.unwrap(); + + // Provoke output. The bash process should echo this back via the PTY. + ws.send(WsMessage::Binary(b"echo hello-broker\n".to_vec())) + .await + .unwrap(); + + let saw_hello = tokio::time::timeout(Duration::from_secs(2), async { + loop { + match ws.next().await { + Some(Ok(WsMessage::Binary(bytes))) => { + if String::from_utf8_lossy(&bytes).contains("hello-broker") { + return true; + } + } + Some(Ok(_)) => continue, + Some(Err(_)) | None => return false, + } + } + }) + .await + .unwrap_or(false); + assert!(saw_hello, "PTY did not echo through the broker WS"); +} + +#[tokio::test] +async fn broker_ws_for_unknown_session_returns_404_not_502() { + let h = spawn("laptop", "naps-laptop", "ws://10.0.0.1:7070").await; + let ws_url = format!("ws://{}/api/attach/{}", h.addr, Uuid::new_v4()); + let err = connect_async(ws_url) + .await + .expect_err("expected handshake fail"); + let msg = err.to_string(); + assert!( + msg.contains("404") || msg.to_lowercase().contains("not found"), + "{msg}" + ); +}