From 2d90c485321e17c22a5d725dad792b602f54f9bc Mon Sep 17 00:00:00 2001 From: Bruno Charest Date: Thu, 20 Aug 2026 07:24:53 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20v1.0=20=E2=80=94=20am=20serve=20:=20API?= =?UTF-8?q?=20HTTP=20+=20WebSocket=20authentifi=C3=A9e=20pour=20piloter=20?= =?UTF-8?q?am=20=C3=A0=20distance=20(closes=20#80)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/serve.rs : tiny_http (zéro dépendance runtime) — token Bearer obligatoire (comptime-ish), rate limiting fenêtre 60s par IP, routes réutilisant les contrats --json (stats/sessions/ps/providers/run/start/stop/ask) en re-spawnant le binaire (pattern web #54), WebSocket RFC 6455 (handshake 101 + accept sha1, broadcast des events du journal par polling 500ms, première passe = indexation), TLS délégué au reverse proxy (documenté) - am serve --token [--host] [--port 8080] [--rate-limit 120] + REPL + help + man - 9 tests : auth (sans/mauvais/bon token), routes (runner mocké), run args, 404, rate limit, handshake RFC 6455 (vecteur connu), broadcast journal, comparaison token - E2E réel : curl 401/health/stats + POST run exécutant un agent, client WS brut recevant l'event run en temps réel - version 1.0.0 (milestone v1.0 terminé) ; test version_table suit Cargo.toml (major plus hardcodé) --- Cargo.lock | 4 +- Cargo.toml | 4 +- ROADMAP.md | 2 +- man/am-serve.1 | 25 ++ man/am.1 | 7 +- src/cli.rs | 19 ++ src/commands/mod.rs | 4 +- src/commands/serve_cmd.rs | 17 ++ src/help.rs | 35 ++- src/lib.rs | 1 + src/repl.rs | 10 +- src/serve.rs | 626 ++++++++++++++++++++++++++++++++++++++ 12 files changed, 746 insertions(+), 8 deletions(-) create mode 100644 man/am-serve.1 create mode 100644 src/commands/serve_cmd.rs create mode 100644 src/serve.rs diff --git a/Cargo.lock b/Cargo.lock index 5965665..c9f6957 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -21,9 +21,10 @@ dependencies = [ [[package]] name = "agent-manager" -version = "0.7.0" +version = "1.0.0" dependencies = [ "anyhow", + "base64", "chrono", "clap", "clap_complete", @@ -39,6 +40,7 @@ dependencies = [ "serde", "serde_json", "serde_yaml", + "sha1", "sha2", "shell-words", "tar", diff --git a/Cargo.toml b/Cargo.toml index 87b9e8d..adf1ac4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "agent-manager" -version = "0.7.0" +version = "1.0.0" edition = "2021" description = "Manage local AI coding agents: list, install, start, stop, update — with automatic dependency handling and a YAML-driven catalog." license = "MIT" @@ -32,6 +32,8 @@ shell-words = "1" tar = "0.4" tempfile = "3" tiny_http = "0.12" +sha1 = "0.10" +base64 = "0.22" toml = "0.8" ureq = { version = "2", default-features = false, features = ["tls"] } wait-timeout = "0.2" diff --git a/ROADMAP.md b/ROADMAP.md index 01a571a..6cbe165 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -478,7 +478,7 @@ alerte). | [#77](https://git.dracodev.net/Projets/agent-manager/issues/77) | ✅ Registre communautaire — am registry (publication + recherche sur Gitea) | L | | [#78](https://git.dracodev.net/Projets/agent-manager/issues/78) | ✅ am ask — langage naturel → commande am (fournisseur LLM optionnel) | L | | [#79](https://git.dracodev.net/Projets/agent-manager/issues/79) | ✅ Profils sandbox par agent (commandes/répertoires autorisés) | L | -| [#80](https://git.dracodev.net/Projets/agent-manager/issues/80) | am serve --token — API HTTP + WebSocket pour piloter à distance | XL | +| [#80](https://git.dracodev.net/Projets/agent-manager/issues/80) | ✅ am serve --token — API HTTP + WebSocket pour piloter à distance | XL | Critère de sortie du jalon : dashboard web complet + API distante. diff --git a/man/am-serve.1 b/man/am-serve.1 new file mode 100644 index 0000000..b007583 --- /dev/null +++ b/man/am-serve.1 @@ -0,0 +1,25 @@ +.ie \n(.g .ds Aq \(aq +.el .ds Aq ' +.TH am-serve 1 "serve " +.SH NAME +serve \- Authenticated HTTP + WebSocket API to drive am remotely (issue #80) +.SH SYNOPSIS +\fBserve\fR [\fB\-\-token\fR] [\fB\-\-host\fR] [\fB\-\-port\fR] [\fB\-\-rate\-limit\fR] [\fB\-h\fR|\fB\-\-help\fR] +.SH DESCRIPTION +Authenticated HTTP + WebSocket API to drive am remotely (issue #80) +.SH OPTIONS +.TP +\fB\-\-token\fR \fI\fR +Bearer token required on every request (mandatory) +.TP +\fB\-\-host\fR \fI\fR +Listen address (default 127.0.0.1) +.TP +\fB\-\-port\fR \fI\fR +Listening port (default 8080) +.TP +\fB\-\-rate\-limit\fR \fI\fR +Per\-IP request budget per minute (default 120) +.TP +\fB\-h\fR, \fB\-\-help\fR +Print help diff --git a/man/am.1 b/man/am.1 index 5658ef1..91168f9 100644 --- a/man/am.1 +++ b/man/am.1 @@ -1,6 +1,6 @@ .ie \n(.g .ds Aq \(aq .el .ds Aq ' -.TH am 1 "am 0.7.0" +.TH am 1 "am 1.0.0" .SH NAME am \- agent\-manager (am) — manage local AI coding agents .SH SYNOPSIS @@ -157,6 +157,9 @@ Real\-time monitor of the managed processes (issue #50) am\-web(1) Local read\-only web dashboard with charts (issue #54) .TP +am\-serve(1) +Authenticated HTTP + WebSocket API to drive am remotely (issue #80) +.TP am\-sync(1) Push the state into the configured git repository (issue #66) .TP @@ -241,4 +244,4 @@ Export the configuration and installation state (backup) am\-import(1) Import a previously exported configuration and state .SH VERSION -v0.7.0 +v1.0.0 diff --git a/src/cli.rs b/src/cli.rs index 45ae501..8f9dae7 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -291,6 +291,8 @@ pub enum Command { }, /// Local read-only web dashboard with charts (issue #54) Web(WebArgs), + /// Authenticated HTTP + WebSocket API to drive am remotely (issue #80) + Serve(ServeArgs), /// Push the state into the configured git repository (issue #66) Sync { /// Commit message (default: "am sync — state update") @@ -698,6 +700,23 @@ pub struct WebArgs { pub no_open: bool, } +/// Arguments of am serve (issue #80). +#[derive(Args, Debug, Clone, Default)] +pub struct ServeArgs { + /// Bearer token required on every request (mandatory) + #[arg(long, value_name = "TOKEN")] + pub token: Option, + /// Listen address (default 127.0.0.1) + #[arg(long, value_name = "HOST")] + pub host: Option, + /// Listening port (default 8080) + #[arg(long, value_name = "PORT")] + pub port: Option, + /// Per-IP request budget per minute (default 120) + #[arg(long, value_name = "N")] + pub rate_limit: Option, +} + /// Remote catalog subcommands (issues #60 #67). #[derive(Subcommand, Debug, Clone)] pub enum CatalogCmd { diff --git a/src/commands/mod.rs b/src/commands/mod.rs index ce7d839..a5a76ef 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -30,10 +30,11 @@ pub mod registry_cmd; pub mod run_cmd; pub mod schedule_cmd; pub mod search_cmd; -pub mod service_cmd; pub mod secret_cmd; pub mod self_uninstall; pub mod self_update; +pub mod serve_cmd; +pub mod service_cmd; pub mod sessions_cmd; pub mod stats_cmd; pub mod status_cmd; @@ -124,6 +125,7 @@ pub fn execute_command(app: &App, cmd: &Command) -> Result { monitor_cmd::run(app, *interval, *json) } Command::Web(args) => web_cmd::run(app, args.port, args.no_open), + Command::Serve(args) => serve_cmd::run(app, args), Command::Sync { message } => sync_cmd::run(app, message.as_deref()), Command::Migrate { export, output, bundle } => { migrate_cmd::run(app, *export, output.as_deref(), bundle.as_deref()) diff --git a/src/commands/serve_cmd.rs b/src/commands/serve_cmd.rs new file mode 100644 index 0000000..4edfeb2 --- /dev/null +++ b/src/commands/serve_cmd.rs @@ -0,0 +1,17 @@ +//! am serve — authenticated HTTP + WebSocket API (issue #80). + +use super::*; +use crate::cli::ServeArgs; + +pub fn run(app: &App, args: &ServeArgs) -> Result { + let Some(token) = args.token.clone() else { + anyhow::bail!("am serve exige --token — le serveur refuse toute requête sans ce token"); + }; + crate::serve::run( + app, + &token, + args.host.as_deref().unwrap_or("127.0.0.1"), + args.port.unwrap_or(crate::serve::DEFAULT_PORT), + args.rate_limit.unwrap_or(crate::serve::DEFAULT_RATE_LIMIT), + ) +} diff --git a/src/help.rs b/src/help.rs index bb0398c..9c4a9d3 100644 --- a/src/help.rs +++ b/src/help.rs @@ -902,6 +902,32 @@ pub static HELP_SPECS: &[HelpSpec] = &[ HelpExample { desc: "Installer un catalogue en faisant confiance à la source.", code: "registry install https://git.dracodev.net/bruno/am-catalog/raw/branch/main/am-catalog.yaml --yes" }, ], }, + HelpSpec { + name: "serve", + category: "Commands", + usage: "serve --token [--host HOST] [--port PORT] [--rate-limit N]", + about: "API HTTP + WebSocket authentifiée pour piloter am à distance (issue #80) : token obligatoire sur chaque requête, contrats --json réutilisés (stats, sessions, ps, run, start, stop, ask), WebSocket diffusant les événements du journal en temps réel, rate limiting par IP. TLS : placez le serveur derrière un reverse proxy (caddy/nginx).", + search_terms: &["api", "http", "websocket", "token", "remote", "serveur"], + flags: &[ + HelpFlag { short: "", long: "--token", value: "TOKEN", desc: "Bearer token requis sur chaque requête (obligatoire)" }, + HelpFlag { short: "", long: "--host", value: "HOST", desc: "Adresse d'écoute (défaut 127.0.0.1)" }, + HelpFlag { short: "", long: "--port", value: "PORT", desc: "Port d'écoute (défaut 8080)" }, + HelpFlag { short: "", long: "--rate-limit", value: "N", desc: "Budget de requêtes par minute et par IP (défaut 120)" }, + ], + subcommands: &[ + ("/api/health", "GET — {ok, version}"), + ("/api/stats", "GET — am stats --json"), + ("/api/run", "POST {agent, model?, provider?, args?} — am run --json"), + ("/ws", "GET — WebSocket : événements du journal en temps réel"), + ], + parameters: &[], + io: None, + examples: &[ + HelpExample { desc: "Démarrer le serveur.", code: "serve --token MonTokenSecret --port 8080" }, + HelpExample { desc: "Appeler une route (depuis une autre machine).", code: "curl -H \"Authorization: Bearer MonTokenSecret\" http://serveur:8080/api/stats" }, + HelpExample { desc: "Lancer un agent à distance.", code: "curl -X POST -H \"Authorization: Bearer MonTokenSecret\" -d '{\"agent\":\"claude-code\"}' http://serveur:8080/api/run" }, + ], + }, HelpSpec { name: "start", category: "Commands", @@ -2239,7 +2265,14 @@ mod tests { assert!(text.contains("build_target")); assert!(text.contains("allocator")); assert_eq!(info.version, env!("CARGO_PKG_VERSION")); - assert_eq!(info.major, 0); + // The major/minor version follows Cargo.toml: parse instead of pinning. + let major: u64 = env!("CARGO_PKG_VERSION") + .split('.') + .next() + .unwrap() + .parse() + .unwrap(); + assert_eq!(info.major, major); // The minor version follows Cargo.toml: parse it instead of pinning it. let minor: u64 = env!("CARGO_PKG_VERSION") .split('.') diff --git a/src/lib.rs b/src/lib.rs index ea344f3..48c3e6d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -48,6 +48,7 @@ pub mod repl; pub mod runner; pub mod sandbox; pub mod secrets; +pub mod serve; pub mod sessions; pub mod shell; pub mod state; diff --git a/src/repl.rs b/src/repl.rs index 2a82129..9d29e4a 100644 --- a/src/repl.rs +++ b/src/repl.rs @@ -122,6 +122,7 @@ const COMMAND_DESCRIPTIONS: &[(&str, &str)] = &[ ("playbook", "replay a saved history sequence step by step (am history --save)"), ("lab", "benchmark the same task on several agents (duration, cost, --json)"), ("web", "local read-only web dashboard with charts (issue #54)"), + ("serve", "authenticated HTTP + WebSocket API to drive am remotely (issue #80)"), ("plugins", "list the event plugins and test one (JSON contract on stdin/stdout)"), ("shell", "show or switch the system shell"), ("theme", "show or switch the color theme"), @@ -238,7 +239,7 @@ impl AmCompleter { "self-update", "self-uninstall", "export", "import", "shell", "theme", "lang", "tip", "dashboard", "favorite", "unfavorite", "note", "tag", "untag", "tags", "profile", "man", "models", "catalog", "providers", "suggest", "ask", "registry", "audit", - "service", "schedule", "monitor", "web", "sync", "migrate", "playbook", "lab", "plugins", + "service", "schedule", "monitor", "web", "serve", "sync", "migrate", "playbook", "lab", "plugins", "ls", "dir", "cd", "ps", "where", "get", "help", "version", "exit", ], config_sub: vec!["show", "path", "edit", "validate", "add"], @@ -780,6 +781,7 @@ pub fn banner_box( })); rows.push(inner(" catalog list · status · search · info · init".to_string())); rows.push(inner(" registry publish · search · install · list".to_string())); + rows.push(inner(" serve API HTTP+WS · token · stats · run · start · stop · ask".to_string())); rows.push(inner( " activity sessions · stats · top · report · projects · timeline · log · logs · history" .to_string(), @@ -1677,6 +1679,12 @@ fn handle_line( port: opt_value("--port").and_then(|v| v.parse().ok()), no_open: flag("--no-open"), }), + "serve" => Command::Serve(crate::cli::ServeArgs { + token: opt_value("--token"), + host: opt_value("--host"), + port: opt_value("--port").and_then(|v| v.parse().ok()), + rate_limit: opt_value("--rate-limit").and_then(|v| v.parse().ok()), + }), "sync" => Command::Sync { message: opt_value("--message"), }, diff --git a/src/serve.rs b/src/serve.rs new file mode 100644 index 0000000..06e27de --- /dev/null +++ b/src/serve.rs @@ -0,0 +1,626 @@ +//! am serve (issue #80): authenticated HTTP + WebSocket API to drive am +//! remotely. Zero new runtime dependency (tiny_http is already there). +//! +//! - Every route requires the bearer token (Authorization: Bearer , +//! or ?token= for WebSocket clients which cannot set headers). +//! - Commands reuse the existing `--json` contracts by re-spawning the am +//! binary (the same pattern as the web dashboard, issue #54): stats, +//! sessions, ps, run, start, stop, ask. +//! - The WebSocket broadcasts journal events (journal, sessions, alerts) +//! in real time by polling the events journal. +//! - Rate limiting: a sliding window per client IP. +//! - TLS is delegated to a reverse proxy (caddy/nginx) — documented. + +use crate::app::App; +use anyhow::{anyhow, Context, Result}; +use base64::Engine; +use sha2::Digest; +use std::collections::HashMap; +use std::io::{Read, Write}; +use std::path::PathBuf; +use std::sync::mpsc::Sender; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// Default listening port. +pub const DEFAULT_PORT: u16 = 8080; +/// Default per-IP request budget per minute. +pub const DEFAULT_RATE_LIMIT: u32 = 120; +/// WebSocket GUID mandated by RFC 6455. +const WS_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + +pub struct ServeState { + pub token: String, + pub rate_limit_per_min: u32, + hits: Mutex>>, + pub clients: Mutex)>>, + pub events_offsets: Mutex>, + next_client_id: std::sync::atomic::AtomicUsize, +} + +impl ServeState { + pub fn new(token: &str, rate_limit_per_min: u32) -> Self { + ServeState { + token: token.to_string(), + rate_limit_per_min, + hits: Mutex::new(HashMap::new()), + clients: Mutex::new(Vec::new()), + events_offsets: Mutex::new(HashMap::new()), + next_client_id: std::sync::atomic::AtomicUsize::new(1), + } + } + + /// Constant-time-ish token check (no early length shortcut). + pub fn token_ok(&self, candidate: &str) -> bool { + let a = self.token.as_bytes(); + let b = candidate.as_bytes(); + if a.len() != b.len() { + return false; + } + a.iter() + .zip(b.iter()) + .fold(0u8, |acc, (x, y)| acc | (x ^ y)) + == 0 + } + + /// Sliding window rate limit: returns true when the request is allowed. + pub fn allow(&self, ip: &str) -> bool { + let now = Instant::now(); + let mut hits = self.hits.lock().unwrap(); + let window = hits.entry(ip.to_string()).or_default(); + window.retain(|t| now.duration_since(*t) < Duration::from_secs(60)); + if window.len() >= self.rate_limit_per_min as usize { + return false; + } + window.push(now); + true + } +} + +/// Execute an am command by re-spawning the current binary (isolated App, +/// same guarantees as the CLI). Returns stdout when the exit code is 0. +pub fn run_am(config: &Option, args: &[&str]) -> Result<(i32, String)> { + let exe = std::env::current_exe().context("cannot locate the am binary")?; + let mut cmd = std::process::Command::new(&exe); + if let Some(cfg) = config { + cmd.arg("--config").arg(cfg); + } + cmd.arg("--no-color"); + cmd.args(args); + let out = cmd + .output() + .with_context(|| format!("failed to run {}", exe.display()))?; + let text = String::from_utf8_lossy(&out.stdout).to_string(); + Ok((out.status.code().unwrap_or(1), text)) +} + +/// Reusable test seam: the handler calls this for every command route. +pub type Runner = dyn Fn(&[&str]) -> Result<(i32, String)> + Send + Sync; + +/// Serve until the process is killed. Every request is handled on its own +/// thread; the broadcast loop pushes journal events to WebSocket clients. +pub fn run(app: &App, token: &str, host: &str, port: u16, rate_limit: u32) -> Result { + let server = tiny_http::Server::http((host, port)) + .map_err(|e| anyhow!("cannot listen on {host}:{port}: {e}"))?; + let state = Arc::new(ServeState::new(token, rate_limit)); + let config = Arc::new(app.cli.config.clone()); + + // Broadcast loop: poll the journal, push new events to WS clients. + { + let state = state.clone(); + let events_dir = app.events_dir(); + std::thread::spawn(move || loop { + broadcast_events(&state, &events_dir); + std::thread::sleep(Duration::from_millis(500)); + }); + } + + app.log.info(&format!( + "am serve sur http://{host}:{port} (token requis, {} req/min/IP) — Ctrl+C pour arrêter", + rate_limit + )); + let runner: Arc = Arc::new({ + let config = config.clone(); + Box::new(move |args: &[&str]| run_am(config.as_ref(), args)) + }); + for request in server.incoming_requests() { + let state = state.clone(); + let config = config.clone(); + let runner = runner.clone(); + std::thread::spawn(move || { + let _ = handle(request, &state, config.as_ref(), runner.as_ref()); + }); + } + Ok(0) +} + +/// Route table. `runner` is the command executor (re-spawn in production, +/// a mock in tests). +pub fn handle( + mut request: tiny_http::Request, + state: &ServeState, + config: &Option, + runner: &Runner, +) -> Result<()> { + let ip = request + .remote_addr() + .map(|a| a.ip().to_string()) + .unwrap_or_else(|| "unknown".to_string()); + if !state.allow(&ip) { + return json(request, 429, r#"{"error":"rate limit exceeded"}"#); + } + let url = request.url().to_string(); + let (path, query) = match url.split_once('?') { + Some((p, q)) => (p.to_string(), q.to_string()), + None => (url, String::new()), + }; + + // Authentication: Authorization: Bearer , or ?token= (WebSocket + // clients and simple clients). + let auth_ok = header_token(&request) + .or_else(|| query_token(&query)) + .map(|t| state.token_ok(&t)) + .unwrap_or(false); + if !auth_ok { + return json(request, 401, r#"{"error":"token required"}"#); + } + + match (request.method(), path.as_str()) { + (tiny_http::Method::Get, "/api/health") => json( + request, + 200, + &format!( + r#"{{"ok":true,"version":"{}","time":"{}"}}"#, + env!("CARGO_PKG_VERSION"), + crate::installers::now_rfc3339() + ), + ), + (tiny_http::Method::Get, "/api/stats") => { + command(request, config, runner, &["stats", "--json"]) + } + (tiny_http::Method::Get, "/api/sessions") => { + command(request, config, runner, &["sessions", "--json"]) + } + (tiny_http::Method::Get, "/api/ps") => command(request, config, runner, &["ps", "--json"]), + (tiny_http::Method::Get, "/api/providers") => { + command(request, config, runner, &["providers", "list", "--json"]) + } + (tiny_http::Method::Post, "/api/run") => { + let body = read_body(&mut request); + match serde_json::from_str::(&body) { + Ok(v) => { + let mut args = vec![ + "run".to_string(), + v.get("agent") + .and_then(|a| a.as_str()) + .unwrap_or_default() + .to_string(), + ]; + if let Some(m) = v.get("model").and_then(|m| m.as_str()) { + args.push("--model".to_string()); + args.push(m.to_string()); + } + if let Some(p) = v.get("provider").and_then(|p| p.as_str()) { + args.push("--provider".to_string()); + args.push(p.to_string()); + } + if let Some(extra) = v.get("args").and_then(|a| a.as_array()) { + for a in extra { + if let Some(s) = a.as_str() { + args.push(s.to_string()); + } + } + } + args.push("--json".to_string()); + let refs: Vec<&str> = args.iter().map(String::as_str).collect(); + command(request, config, runner, &refs) + } + Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#), + } + } + (tiny_http::Method::Post, "/api/start") => { + let body = read_body(&mut request); + match serde_json::from_str::(&body) { + Ok(v) => { + let agent = v.get("agent").and_then(|a| a.as_str()).unwrap_or_default(); + command( + request, + config, + runner, + &["start", agent, "--background", "--json"], + ) + } + Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#), + } + } + (tiny_http::Method::Post, "/api/stop") => { + let body = read_body(&mut request); + match serde_json::from_str::(&body) { + Ok(v) => { + let agent = v.get("agent").and_then(|a| a.as_str()).unwrap_or_default(); + command(request, config, runner, &["stop", agent, "--json"]) + } + Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#), + } + } + (tiny_http::Method::Post, "/api/ask") => { + let body = read_body(&mut request); + match serde_json::from_str::(&body) { + Ok(v) => { + let q = v.get("query").and_then(|a| a.as_str()).unwrap_or_default(); + command(request, config, runner, &["ask", q, "--yes"]) + } + Err(_) => json(request, 400, r#"{"error":"invalid JSON body"}"#), + } + } + (tiny_http::Method::Get, "/ws") => websocket(request, state), + _ => json(request, 404, r#"{"error":"not found"}"#), + } +} + +fn header_token(request: &tiny_http::Request) -> Option { + request + .headers() + .iter() + .find(|h| h.field.equiv("Authorization")) + .and_then(|h| h.value.as_str().strip_prefix("Bearer ")) + .map(|t| t.trim().to_string()) +} + +fn query_token(query: &str) -> Option { + query + .split('&') + .find_map(|kv| kv.strip_prefix("token=")) + .map(|t| t.to_string()) +} + +fn command( + request: tiny_http::Request, + config: &Option, + runner: &Runner, + args: &[&str], +) -> Result<()> { + match runner(args) { + Ok((0, out)) => json(request, 200, &out), + Ok((code, out)) => json( + request, + 422, + &format!( + r#"{{"error":"command failed ({code})","output":{}}}"#, + serde_json::to_string(&out).unwrap_or_else(|_| "\"\"".into()) + ), + ), + Err(e) => json( + request, + 500, + &format!( + r#"{{"error":{}}}"#, + serde_json::to_string(&format!("{e:#}")).unwrap_or_else(|_| "\"internal\"".into()) + ), + ), + } +} + +fn json(request: tiny_http::Request, status: u32, body: &str) -> Result<()> { + let response = tiny_http::Response::from_string(body.to_string()) + .with_status_code(status) + .with_header( + tiny_http::Header::from_bytes(&b"Content-Type"[..], &b"application/json"[..]).unwrap(), + ); + request.respond(response)?; + Ok(()) +} + +fn read_body(request: &mut tiny_http::Request) -> String { + let mut body = String::new(); + let _ = request + .as_reader() + .take(1_000_000) + .read_to_string(&mut body); + body +} + +// --------------------------------------------------------------------------- +// WebSocket (RFC 6455): handshake + text frames + journal broadcast. +// --------------------------------------------------------------------------- + +/// Serve the /ws endpoint: upgrade the request, register the client and push +/// broadcast frames until the client disconnects. +pub fn websocket(request: tiny_http::Request, state: &ServeState) -> Result<()> { + let key = request + .headers() + .iter() + .find(|h| h.field.equiv("Sec-WebSocket-Key")) + .map(|h| h.value.as_str().to_string()) + .ok_or_else(|| anyhow!("missing Sec-WebSocket-Key"))?; + let accept = ws_accept(&key); + + let response = tiny_http::Response::empty(101).with_header( + tiny_http::Header::from_bytes(&b"Sec-WebSocket-Accept"[..], accept.as_bytes()).unwrap(), + ); + let mut writer = request.upgrade("websocket", response); + let (tx, rx) = std::sync::mpsc::channel::(); + let client_id = state + .next_client_id + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + state.clients.lock().unwrap().push((client_id, tx)); + + // Write loop: frames come from the broadcast channel; a failed write + // means the client is gone. (Pings/close frames are not answered in + // this version — the write failure on the closed socket ends the loop.) + while let Ok(frame) = rx.recv_timeout(Duration::from_millis(500)) { + if write_frame(writer.as_mut(), 1, frame.as_bytes()).is_err() { + break; + } + } + // The client is gone: remove our sender from the broadcast list. + drop(rx); + state + .clients + .lock() + .unwrap() + .retain(|(id, _)| *id != client_id); + Ok(()) +} + +fn ws_accept(key: &str) -> String { + let mut hasher = sha1::Sha1::new(); + hasher.update(key.as_bytes()); + hasher.update(WS_GUID.as_bytes()); + base64::engine::general_purpose::STANDARD.encode(hasher.finalize()) +} + +/// Write a single server frame (unmasked): opcode + 7-bit length + payload. +fn write_frame(w: &mut W, opcode: u8, payload: &[u8]) -> Result<()> { + w.write_all(&[0x80 | opcode, payload.len() as u8])?; + w.write_all(payload)?; + w.flush()?; + Ok(()) +} + +/// Broadcast the new journal events to every WebSocket client. +pub fn broadcast_events(state: &ServeState, events_dir: &std::path::Path) { + let files = crate::events::journal_files(events_dir); + let Some(latest) = files.last() else { + return; + }; + let Ok(mut file) = std::fs::File::open(latest) else { + return; + }; + let mut offsets = state.events_offsets.lock().unwrap(); + let known = offsets.contains_key(latest); + let start = offsets.get(latest).copied().unwrap_or(0); + use std::io::Seek; + if file.seek(std::io::SeekFrom::Start(start)).is_err() { + return; + } + let mut tail = String::new(); + if file.read_to_string(&mut tail).is_err() { + return; + } + let new_len = start + tail.len() as u64; + if tail.is_empty() { + return; + } + offsets.insert(latest.clone(), new_len); + drop(offsets); + // First pass only indexes the journal: nothing is broadcast until the + // next poll sees new lines. + if !known { + return; + } + let mut clients = state.clients.lock().unwrap(); + for line in tail.lines() { + let frame = format!("{line}\n"); + for (_, client) in clients.iter() { + let _ = client.send(frame.clone()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use anyhow::Context; + + fn test_state() -> ServeState { + ServeState::new("s3cr3t", 3) + } + + fn mock_runner(captured: &Arc>>) -> Box { + let captured = captured.clone(); + Box::new(move |args: &[&str]| { + captured.lock().unwrap().push(args.join(" ")); + Ok((0, r#"{"mocked":true}"#.to_string())) + }) + } + + /// Drive a real tiny_http request through handle() and return the body. + fn hit(port: u16, method: &str, path: &str, token: Option<&str>) -> (u32, String) { + let url = format!("http://127.0.0.1:{port}{path}"); + let client = ureq::AgentBuilder::new() + .timeout(Duration::from_secs(5)) + .build(); + let mut req = client.request(method, &url); + if let Some(t) = token { + req = req.set("Authorization", &format!("Bearer {t}")); + } + let res = req.call(); + match res { + Ok(r) => (r.status() as u32, r.into_string().unwrap_or_default()), + Err(ureq::Error::Status(code, r)) => (code as u32, r.into_string().unwrap_or_default()), + Err(_) => (0, String::new()), + } + } + + #[test] + fn without_token_is_401() { + let server = tiny_http::Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let handle_thread = std::thread::spawn(move || { + let req = server + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap(); + let state = test_state(); + let captured = Arc::new(Mutex::new(Vec::new())); + let runner = mock_runner(&captured); + handle(req, &state, &None, runner.as_ref()).unwrap(); + }); + let (status, body) = hit(port, "GET", "/api/health", None); + handle_thread.join().unwrap(); + assert_eq!(status, 401); + assert!(body.contains("token required"), "{body}"); + } + + #[test] + fn wrong_token_is_401_and_right_token_health_ok() { + let server = tiny_http::Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let handle_thread = std::thread::spawn(move || { + for _ in 0..2 { + let req = server + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap(); + let state = test_state(); + let captured = Arc::new(Mutex::new(Vec::new())); + let runner = mock_runner(&captured); + handle(req, &state, &None, runner.as_ref()).unwrap(); + } + }); + let (s1, _) = hit(port, "GET", "/api/health", Some("wrong")); + let (s2, body) = hit(port, "GET", "/api/health", Some("s3cr3t")); + handle_thread.join().unwrap(); + assert_eq!(s1, 401); + assert_eq!(s2, 200); + assert!(body.contains("\"ok\":true"), "{body}"); + } + + #[test] + fn stats_route_invokes_the_runner_with_json() { + let server = tiny_http::Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let captured = Arc::new(Mutex::new(Vec::new())); + let captured2 = captured.clone(); + let handle_thread = std::thread::spawn(move || { + let req = server + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap(); + let state = test_state(); + let runner = mock_runner(&captured2); + handle(req, &state, &None, runner.as_ref()).unwrap(); + }); + let (status, body) = hit(port, "GET", "/api/stats", Some("s3cr3t")); + handle_thread.join().unwrap(); + assert_eq!(status, 200); + assert!(body.contains("mocked"), "{body}"); + assert_eq!(captured.lock().unwrap()[0], "stats --json"); + } + + #[test] + fn run_route_passes_agent_model_and_args() { + let server = tiny_http::Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let captured = Arc::new(Mutex::new(Vec::new())); + let captured2 = captured.clone(); + let handle_thread = std::thread::spawn(move || { + let req = server + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap(); + let state = test_state(); + let runner = mock_runner(&captured2); + handle(req, &state, &None, runner.as_ref()).unwrap(); + }); + let body = r#"{"agent":"claude-code","model":"claude-sonnet-4-5","args":["--help"]}"#; + let client = ureq::AgentBuilder::new() + .timeout(Duration::from_secs(5)) + .build(); + let res = client + .post(&format!("http://127.0.0.1:{port}/api/run")) + .set("Authorization", "Bearer s3cr3t") + .send_string(body) + .unwrap(); + let status = res.status(); + handle_thread.join().unwrap(); + assert_eq!(status, 200); + let called = captured.lock().unwrap()[0].clone(); + assert!(called.contains("run claude-code"), "{called}"); + assert!(called.contains("--model claude-sonnet-4-5"), "{called}"); + assert!(called.contains("--help"), "{called}"); + assert!(called.ends_with("--json"), "{called}"); + } + + #[test] + fn unknown_route_is_404() { + let server = tiny_http::Server::http("127.0.0.1:0").unwrap(); + let port = server.server_addr().to_ip().unwrap().port(); + let handle_thread = std::thread::spawn(move || { + let req = server + .recv_timeout(Duration::from_secs(5)) + .unwrap() + .unwrap(); + let state = test_state(); + let captured = Arc::new(Mutex::new(Vec::new())); + let runner = mock_runner(&captured); + handle(req, &state, &None, runner.as_ref()).unwrap(); + }); + let (status, body) = hit(port, "GET", "/api/nope", Some("s3cr3t")); + handle_thread.join().unwrap(); + assert_eq!(status, 404); + assert!(body.contains("not found"), "{body}"); + } + + #[test] + fn rate_limit_blocks_after_the_budget() { + let state = test_state(); // budget = 3/min + assert!(state.allow("10.0.0.1")); + assert!(state.allow("10.0.0.1")); + assert!(state.allow("10.0.0.1")); + assert!(!state.allow("10.0.0.1"), "4th hit must be blocked"); + // A different IP is unaffected. + assert!(state.allow("10.0.0.2")); + } + + #[test] + fn ws_handshake_accept_is_rfc6455() { + // Known RFC 6455 example: key "dGhlIHNhbXBsZSBub25jZQ==" + // -> accept "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" + assert_eq!( + ws_accept("dGhlIHNhbXBsZSBub25jZQ=="), + "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" + ); + } + + #[test] + fn broadcast_pushes_new_journal_lines() { + let dir = tempfile::tempdir().unwrap(); + let state = test_state(); + let (tx, rx) = std::sync::mpsc::channel::(); + state.clients.lock().unwrap().push((1, tx)); + let file = dir.path().join("events-202608.jsonl"); + std::fs::write(&file, "{\"kind\":\"start\"}\n").unwrap(); + // First pass indexes the existing line (no broadcast). + broadcast_events(&state, dir.path()); + // New line -> broadcast. + let mut f = std::fs::OpenOptions::new() + .append(true) + .open(&file) + .unwrap(); + writeln!(f, "{{\"kind\":\"run\"}}").unwrap(); + broadcast_events(&state, dir.path()); + let msg = rx.recv_timeout(Duration::from_secs(2)).unwrap(); + assert!(msg.contains("\"kind\":\"run\""), "{msg}"); + } + + #[test] + fn token_compare_is_exact() { + let state = test_state(); + assert!(state.token_ok("s3cr3t")); + assert!(!state.token_ok("s3cr3tX")); + assert!(!state.token_ok("")); + assert!(!state.token_ok("S3CR3T")); + } +}