Skip to main content

cellos_ctl/cmd/
webui.rs

1//! `cellctl webui` — localhost reverse proxy + static bundle host.
2//!
3//! Per ADR-0017, the web view ships *with cellctl*, not with cellos-server.
4//! Operators invoke `cellctl webui` to spin up a foreground localhost proxy:
5//!
6//!   - Serves `crates/cellos-ctl/static/` (the Vite build output) at `/`.
7//!   - Reverse-proxies `GET /v1/*` and `GET /ws/events` upstream to the
8//!     cellos-server URL configured in `~/.cellctl/config`, injecting the
9//!     bearer token on the way out. The bundle never sees the token.
10//!   - Refuses any non-`GET` method with HTTP 405 (`Allow: GET`). This is the
11//!     structural enforcement of ADR-0016's read-only browser boundary.
12//!   - Binds a session token in a URL fragment (`/#sess=<base64>`) and
13//!     swaps it for an `HttpOnly; SameSite=Strict` cookie via
14//!     `POST /auth/exchange`. Subsequent proxy + WS requests require the
15//!     cookie. The fragment is cleared by the bundle's bootstrap once
16//!     the cookie is set.
17//!   - Exits cleanly on SIGINT.
18//!
19//! Bind modes (ADR-0017 §Decision 4):
20//!
21//! - `--bind auto` (default, on Unix): bind BOTH a loopback TCP port (for the
22//!   browser) AND a Unix socket at
23//!   `${XDG_RUNTIME_DIR:-/tmp}/cellctl-webui-<pid>.sock` (for inter-process
24//!   tooling that wants to bypass loopback). On Windows, `auto` degrades to
25//!   loopback-only.
26//! - `--bind loopback`: TCP loopback only.
27//! - `--bind unix` (Unix only): Unix socket only. The browser cannot reach a
28//!   Unix socket directly — this mode is for inter-process forwarders (e.g.
29//!   `socat` / `ssh -L`). On Windows this errors out.
30//!
31//! The Unix socket is created with mode `0600` (operator-owned only) and is
32//! removed on graceful shutdown (SIGINT).
33
34use std::net::SocketAddr;
35use std::path::PathBuf;
36use std::sync::Arc;
37
38use axum::body::{Body, Bytes};
39use axum::extract::{ws::WebSocketUpgrade, Path as AxumPath, State};
40use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
41use axum::response::{IntoResponse, Response};
42use axum::routing::{any, post};
43use axum::Router;
44use base64::Engine as _;
45use clap::ValueEnum;
46use futures_util::{SinkExt, StreamExt};
47use serde::Deserialize;
48use tokio::sync::RwLock;
49use tower_http::services::ServeDir;
50
51use crate::client::CellosClient;
52use crate::config::Config;
53use crate::exit::{CtlError, CtlResult};
54
55/// Where the bundle lives on disk. Resolved relative to the cellctl binary's
56/// crate at compile time; we look up the actual path at runtime via the
57/// `CARGO_MANIFEST_DIR` env var (set in tests) or `./static` (when the
58/// binary is installed alongside its bundle).
59const BUNDLE_DIR_RELATIVE: &str = "static";
60
61/// Bind mode for the webui proxy.
62#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
63pub enum BindMode {
64    /// Default: on Unix, bind BOTH a loopback TCP port and a Unix socket.
65    /// On Windows, degrade to loopback-only.
66    #[default]
67    Auto,
68    /// Force 127.0.0.1 on a random high port. No Unix socket.
69    Loopback,
70    /// Force a Unix socket only (no TCP). Browsers cannot reach a Unix
71    /// socket directly; this mode is for inter-process forwarders. Errors
72    /// on Windows.
73    Unix,
74}
75
76/// Per-process shared state: upstream client + the one valid session token
77/// (until it's exchanged) and the resulting session cookie value.
78///
79/// SECURITY NOTE (red-team pass B, HIGH-5): never log `upstream_bearer` or
80/// `session_token` outside this module. The proxy injects `Authorization:
81/// Bearer ...` headers into outbound requests; do NOT enable
82/// `RUST_LOG=reqwest=trace` in production — `reqwest` will dump those
83/// headers to stderr. The audit log path in `cmd/webui.rs` keeps the
84/// startup banner deliberately bearer-free.
85#[derive(Clone)]
86struct AppState {
87    /// Upstream cellos-server base URL (e.g. `http://127.0.0.1:8080`).
88    upstream_base: Arc<String>,
89    /// Upstream Bearer token to inject on every proxied request, if any.
90    upstream_bearer: Arc<Option<String>>,
91    /// The unguessable, single-use session token printed in the URL fragment.
92    session_token: Arc<String>,
93    /// Once `/auth/exchange` succeeds, this holds the cookie value the
94    /// browser is expected to present on every subsequent request.
95    session_cookie: Arc<RwLock<Option<String>>>,
96    /// Single-use gate (red-team pass B, CRIT-1): once `/auth/exchange` has
97    /// been successfully called, every subsequent call MUST return 401
98    /// regardless of whether the provided token matches. Closes the
99    /// "attacker steals the URL fragment, races the operator" path.
100    exchange_consumed: Arc<std::sync::atomic::AtomicBool>,
101    /// Filesystem path of the static bundle (for `ServeDir`).
102    bundle_dir: Arc<PathBuf>,
103}
104
105#[derive(Deserialize)]
106struct ExchangeRequest {
107    sess: String,
108}
109
110/// Entry point invoked by `main.rs`.
111pub async fn run(cfg: &Config, open: bool, bind: BindMode) -> CtlResult<()> {
112    // Reuse the same effective config CellosClient sees, so the URL/token
113    // injected into the proxy matches what the rest of cellctl would use.
114    // Touching CellosClient also validates that the token (if any) parses as
115    // a header value.
116    let _ = CellosClient::new(cfg)?;
117
118    let upstream_base = cfg.effective_server().trim_end_matches('/').to_string();
119    let upstream_bearer = cfg.effective_token();
120
121    let bundle_dir = resolve_bundle_dir()?;
122    if !bundle_dir.join("index.html").exists() {
123        return Err(CtlError::usage(format!(
124            "webui bundle not found at {}/index.html — run `npm --prefix web run build` first",
125            bundle_dir.display()
126        )));
127    }
128
129    let session_token = mint_session_token();
130
131    let upstream_log = upstream_base.clone();
132    let state = AppState {
133        upstream_base: Arc::new(upstream_base),
134        upstream_bearer: Arc::new(upstream_bearer),
135        session_token: Arc::new(session_token.clone()),
136        session_cookie: Arc::new(RwLock::new(None)),
137        exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
138        bundle_dir: Arc::new(bundle_dir.clone()),
139    };
140
141    let app = build_router(state);
142
143    // Decide which listeners we actually want for this run.
144    let (want_tcp, _want_unix) = resolve_bind_plan(bind)?;
145
146    let tcp_listener = if want_tcp {
147        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
148        Some(
149            tokio::net::TcpListener::bind(addr)
150                .await
151                .map_err(|e| CtlError::usage(format!("bind 127.0.0.1: {e}")))?,
152        )
153    } else {
154        None
155    };
156
157    #[cfg(unix)]
158    let unix_socket_path: Option<PathBuf> = if _want_unix {
159        Some(unix_socket_path_for_pid(std::process::id()))
160    } else {
161        None
162    };
163    #[cfg(not(unix))]
164    let unix_socket_path: Option<PathBuf> = None;
165
166    let browser_url = if let Some(l) = tcp_listener.as_ref() {
167        let local_addr = l
168            .local_addr()
169            .map_err(|e| CtlError::usage(format!("local_addr: {e}")))?;
170        Some(format!("http://{}/#sess={}", local_addr, session_token))
171    } else {
172        None
173    };
174
175    if let Some(url) = browser_url.as_ref() {
176        println!("cellctl webui: {}", url);
177    }
178    if let Some(p) = unix_socket_path.as_ref() {
179        println!("cellctl webui: unix://{}", p.display());
180    }
181    if browser_url.is_none() {
182        // unix-only mode: be loud that the browser can't reach this.
183        eprintln!(
184            "cellctl webui: --bind unix has no browser-reachable URL; \
185             use a forwarder (e.g. `socat TCP-LISTEN:0,reuseaddr,fork UNIX-CONNECT:{}`) \
186             or rerun with `--bind auto` for a loopback URL.",
187            unix_socket_path
188                .as_ref()
189                .map(|p| p.display().to_string())
190                .unwrap_or_else(|| "<socket>".to_string())
191        );
192    }
193    // LOW-1 (red-team pass B): print the actual upstream URL the proxy will
194    // forward to, not the dead "(see state)" placeholder. The bearer token
195    // is never printed — only the base URL.
196    println!("upstream: {}", upstream_log);
197    println!("press Ctrl-C to stop");
198
199    if open {
200        if let Some(url) = browser_url.as_ref() {
201            // MED-4 (red-team pass B): belt-and-suspenders sanity check on
202            // the URL before handing it to `opener` (which shells out to
203            // `xdg-open` / `open` / `Start-Process`). Today the URL is built
204            // from `local_addr` + a base64-url-no-pad token — both
205            // controlled — but a future contributor might add a header
206            // param that smuggles control characters. Reject anything that
207            // isn't a clean loopback http URL.
208            if !is_safe_open_url(url) {
209                eprintln!(
210                    "cellctl webui: refusing to open URL (failed loopback-http sanity check)"
211                );
212            } else if let Err(e) = opener::open(url) {
213                eprintln!("cellctl webui: could not launch browser: {e}");
214            }
215        } else {
216            eprintln!("cellctl webui: --open ignored: no loopback URL bound (use --bind auto)");
217        }
218    }
219
220    // Build a shutdown signal future shared by both listeners. We resolve on
221    // SIGINT (Ctrl-C) and SIGTERM (kubernetes / systemd / `kill <pid>`) and
222    // (best-effort) clean up the socket file before either listener unbinds
223    // it. HIGH-6 (red-team pass B): without the SIGTERM handler, container
224    // orchestrators hard-kill the process and leave a stale Unix socket
225    // file behind for the next run to trip over.
226    let (shutdown_tx, _shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
227    {
228        let shutdown_tx = shutdown_tx.clone();
229        tokio::spawn(async move {
230            wait_for_shutdown_signal().await;
231            eprintln!("shutting down");
232            let _ = shutdown_tx.send(());
233        });
234    }
235
236    // Spawn the TCP server if requested.
237    let tcp_task = if let Some(listener) = tcp_listener {
238        let app = app.clone();
239        let mut rx = shutdown_tx.subscribe();
240        Some(tokio::spawn(async move {
241            axum::serve(listener, app)
242                .with_graceful_shutdown(async move {
243                    let _ = rx.recv().await;
244                })
245                .await
246        }))
247    } else {
248        None
249    };
250
251    // Spawn the Unix server if requested.
252    #[cfg(unix)]
253    let unix_task = if let Some(path) = unix_socket_path.clone() {
254        let app = app.clone();
255        let mut rx = shutdown_tx.subscribe();
256        let listener = bind_unix_listener(&path)?;
257        Some(tokio::spawn(async move {
258            serve_unix(listener, app, async move {
259                let _ = rx.recv().await;
260            })
261            .await
262        }))
263    } else {
264        None
265    };
266    #[cfg(not(unix))]
267    let unix_task: Option<tokio::task::JoinHandle<std::io::Result<()>>> = None;
268
269    // Wait for both servers (whichever exist). Capture the first error.
270    let mut first_err: Option<String> = None;
271    if let Some(t) = tcp_task {
272        match t.await {
273            Ok(Ok(())) => {}
274            Ok(Err(e)) => {
275                let _ = first_err.get_or_insert_with(|| format!("tcp: {e}"));
276            }
277            Err(e) => {
278                let _ = first_err.get_or_insert_with(|| format!("tcp join: {e}"));
279            }
280        }
281    }
282    if let Some(t) = unix_task {
283        match t.await {
284            Ok(Ok(())) => {}
285            Ok(Err(e)) => {
286                let _ = first_err.get_or_insert_with(|| format!("unix: {e}"));
287            }
288            Err(e) => {
289                let _ = first_err.get_or_insert_with(|| format!("unix join: {e}"));
290            }
291        }
292    }
293
294    // Best-effort cleanup: remove the socket file. (UnixListener does not unlink
295    // on drop.)
296    if let Some(p) = unix_socket_path.as_ref() {
297        let _ = std::fs::remove_file(p);
298    }
299
300    if let Some(e) = first_err {
301        return Err(CtlError::api(format!("webui server: {e}")));
302    }
303    Ok(())
304}
305
306/// Decide which listeners to spawn for a given bind mode.
307///
308/// Returns `(want_tcp, want_unix)`. On non-Unix platforms, `want_unix` is
309/// always false and `BindMode::Unix` errors.
310fn resolve_bind_plan(bind: BindMode) -> CtlResult<(bool, bool)> {
311    #[cfg(unix)]
312    {
313        Ok(match bind {
314            BindMode::Auto => (true, true),
315            BindMode::Loopback => (true, false),
316            BindMode::Unix => (false, true),
317        })
318    }
319    #[cfg(not(unix))]
320    {
321        Ok(match bind {
322            BindMode::Auto => (true, false),
323            BindMode::Loopback => (true, false),
324            BindMode::Unix => {
325                return Err(CtlError::usage(
326                    "--bind unix is not supported on Windows; use --bind loopback (the default)",
327                ));
328            }
329        })
330    }
331}
332
333/// Compute the Unix socket path for a given pid.
334///
335/// Preferred parent: `$XDG_RUNTIME_DIR` (operator-owned, mode 0700 on most
336/// distros). Fallback: `/tmp/cellctl-webui-<uid>/` — a per-uid subdirectory
337/// we create with mode 0700 so the world-writable sticky `/tmp` cannot host
338/// the socket directly. HIGH-2 (red-team pass B): the previous "drop the
339/// socket straight into /tmp" fallback was vulnerable to a same-host
340/// attacker with write access to `/tmp` swapping symlinks under the chmod
341/// call.
342#[cfg(unix)]
343fn unix_socket_path_for_pid(pid: u32) -> PathBuf {
344    let dir = if let Some(xdg) = std::env::var_os("XDG_RUNTIME_DIR") {
345        PathBuf::from(xdg)
346    } else {
347        // Fallback: per-uid sub-directory in /tmp so we never bind directly
348        // under a world-writable parent.
349        let uid = unsafe { libc::getuid() };
350        PathBuf::from("/tmp").join(format!("cellctl-{uid}"))
351    };
352    dir.join(format!("cellctl-webui-{pid}.sock"))
353}
354
355/// Bind a Unix socket at `path` with mode 0600.
356///
357/// HIGH-2 (red-team pass B): the previous implementation called
358/// `set_permissions(path, 0600)` AFTER `bind`, which follows symlinks — an
359/// attacker who can write to the parent directory could symlink-swap the
360/// socket and trick us into chmodding an attacker-chosen target. We close
361/// the window two ways:
362///
363///   1. Ensure the parent directory exists with mode 0700 (operator-only).
364///      If `$XDG_RUNTIME_DIR` is unset, we use a per-uid sub-directory of
365///      `/tmp` that we create ourselves rather than binding directly into
366///      world-writable `/tmp`.
367///   2. Set `umask(0o077)` immediately before `bind` and restore it after.
368///      `UnixListener::bind` creates the socket file with permissions
369///      `0o666 & !umask` — with umask 0o077 we land on 0o600 atomically,
370///      so there is no observable window where the socket exists with
371///      broader permissions. The trailing `set_permissions` becomes a
372///      belt-and-suspenders idempotent.
373#[cfg(unix)]
374fn bind_unix_listener(path: &std::path::Path) -> CtlResult<tokio::net::UnixListener> {
375    use std::os::unix::fs::PermissionsExt;
376
377    // Ensure parent directory exists and is 0700.
378    if let Some(parent) = path.parent() {
379        if !parent.exists() {
380            std::fs::create_dir_all(parent)
381                .map_err(|e| CtlError::usage(format!("mkdir {}: {e}", parent.display())))?;
382        }
383        // Tighten to 0700 unconditionally — if someone else owns the parent
384        // with broader perms, set_permissions will return EPERM, which we
385        // surface to the operator.
386        let _ = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700));
387    }
388
389    // Remove a stale socket file from a prior crashed run. Best-effort.
390    let _ = std::fs::remove_file(path);
391
392    // Atomic mode-0600 bind via tight umask. SAFETY: umask is a process-
393    // global setting; we hold it tight only across the bind call and
394    // restore it immediately. The restore is essential — without it, any
395    // file the rest of the process creates inherits 0600, which surprises
396    // future contributors and breaks tracing-subscriber log file output.
397    let prev_umask = unsafe { libc::umask(0o077) };
398    let bind_result = tokio::net::UnixListener::bind(path);
399    unsafe {
400        let _ = libc::umask(prev_umask);
401    }
402    let listener =
403        bind_result.map_err(|e| CtlError::usage(format!("bind {}: {e}", path.display())))?;
404
405    // Belt-and-suspenders: ensure final mode is 0600. NOTE: this still
406    // follows symlinks, but the umask-driven atomic bind above already
407    // produced the right mode; this call exists to catch the edge case
408    // where the platform's bind() ignored umask for some reason.
409    let perms = std::fs::Permissions::from_mode(0o600);
410    std::fs::set_permissions(path, perms)
411        .map_err(|e| CtlError::usage(format!("chmod 0600 {}: {e}", path.display())))?;
412
413    Ok(listener)
414}
415
416/// HIGH-6 (red-team pass B): wait for either SIGINT (Ctrl-C) or, on Unix,
417/// SIGTERM. On Windows, only Ctrl-C is observable; SIGTERM doesn't exist.
418async fn wait_for_shutdown_signal() {
419    #[cfg(unix)]
420    {
421        use tokio::signal::unix::{signal, SignalKind};
422        let mut sigint = match signal(SignalKind::interrupt()) {
423            Ok(s) => s,
424            Err(_) => {
425                let _ = tokio::signal::ctrl_c().await;
426                return;
427            }
428        };
429        let mut sigterm = match signal(SignalKind::terminate()) {
430            Ok(s) => s,
431            Err(_) => {
432                let _ = sigint.recv().await;
433                return;
434            }
435        };
436        tokio::select! {
437            _ = sigint.recv() => {},
438            _ = sigterm.recv() => {},
439        }
440    }
441    #[cfg(not(unix))]
442    {
443        let _ = tokio::signal::ctrl_c().await;
444    }
445}
446
447/// MED-4 (red-team pass B): sanity check the URL we're about to hand to
448/// `opener::open`. Belt-and-suspenders — today this is always a clean
449/// loopback http URL, but a future contributor adding a query param could
450/// smuggle control characters into a shell-out. Reject anything that:
451///
452///   - doesn't parse as a `url::Url`
453///   - isn't `http` or `https`
454///   - resolves to a non-loopback host
455///   - contains any ASCII control character anywhere in the URL string
456fn is_safe_open_url(s: &str) -> bool {
457    // Any control character (incl. CR, LF, NUL, escape) is an immediate
458    // reject — these can shell-escape on some platforms.
459    if s.chars().any(|c| c.is_control()) {
460        return false;
461    }
462    let Ok(u) = url::Url::parse(s) else {
463        return false;
464    };
465    let scheme = u.scheme();
466    if scheme != "http" && scheme != "https" {
467        return false;
468    }
469    let Some(host) = u.host_str() else {
470        return false;
471    };
472    // Only loopback. cellctl webui binds 127.0.0.1 — refuse if the URL
473    // somehow points elsewhere.
474    matches!(host, "127.0.0.1" | "localhost" | "::1" | "[::1]")
475}
476
477/// Serve `app` over a Unix-domain `UnixListener` with graceful shutdown.
478///
479/// `axum::serve` requires a `tokio::net::TcpListener`, so for Unix sockets we
480/// drop down to the lower-level `accept_loop`-style approach using `hyper`
481/// directly via `tower::Service::call`.
482#[cfg(unix)]
483async fn serve_unix(
484    listener: tokio::net::UnixListener,
485    app: Router,
486    shutdown: impl std::future::Future<Output = ()>,
487) -> std::io::Result<()> {
488    use std::convert::Infallible;
489    use tower::Service;
490
491    // axum::Router implements tower::Service<Request<Body>>. We accept Unix
492    // connections and hand each one to a hyper http1 connection driving the
493    // router. This mirrors what axum::serve does internally for TCP, minus the
494    // TCP-specific bits.
495    tokio::pin!(shutdown);
496
497    loop {
498        tokio::select! {
499            _ = &mut shutdown => return Ok(()),
500            accepted = listener.accept() => {
501                let (stream, _peer) = match accepted {
502                    Ok(s) => s,
503                    Err(e) => {
504                        // EMFILE / transient — log and continue, do not crash.
505                        eprintln!("webui: unix accept error: {e}");
506                        continue;
507                    }
508                };
509                let app = app.clone();
510                tokio::spawn(async move {
511                    // `service_fn` requires `Fn`, but Router::call needs
512                    // `&mut self`. Clone the router per request so each
513                    // invocation owns its own mutable handle.
514                    let svc = hyper::service::service_fn(move |req: http::Request<hyper::body::Incoming>| {
515                        let mut router = app.clone();
516                        async move {
517                            let resp: Response = match router.call(req.map(Body::new)).await {
518                                Ok(r) => r,
519                                Err(never) => match never {},
520                            };
521                            Ok::<_, Infallible>(resp)
522                        }
523                    });
524                    let io = hyper_util::rt::TokioIo::new(stream);
525                    let _ = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
526                        .serve_connection_with_upgrades(io, svc)
527                        .await;
528                });
529            }
530        }
531    }
532}
533
534/// Mint a 32-byte random session token and base64-url-encode it.
535fn mint_session_token() -> String {
536    // rand 0.10: `random()` draws from the thread-local CSPRNG (`ThreadRng`,
537    // ChaCha-based). Filling a `[u8; 32]` this way is equivalent to the old
538    // `thread_rng().fill_bytes(&mut buf)` and stays infallible.
539    let buf: [u8; 32] = rand::random();
540    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf)
541}
542
543/// Resolve the bundle directory. Tests set `CELLCTL_WEBUI_BUNDLE_DIR`
544/// explicitly. Installed binaries look next to `crates/cellos-ctl/static/`
545/// via `CARGO_MANIFEST_DIR` (build-time embed) or the current working dir.
546fn resolve_bundle_dir() -> CtlResult<PathBuf> {
547    if let Ok(p) = std::env::var("CELLCTL_WEBUI_BUNDLE_DIR") {
548        return Ok(PathBuf::from(p));
549    }
550    // At build time we record the crate root; installed binaries can still
551    // override via the env var above.
552    let from_manifest = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(BUNDLE_DIR_RELATIVE);
553    Ok(from_manifest)
554}
555
556/// Build the axum router with the static fileserver, /auth/exchange handler,
557/// and the catch-all proxy route.
558fn build_router(state: AppState) -> Router {
559    // Static bundle: ServeDir handles GET only — non-GET on the static tree
560    // falls through to our catch-all 405 layer.
561    let bundle_dir = state.bundle_dir.as_ref().clone();
562    let serve_dir = ServeDir::new(&bundle_dir).append_index_html_on_directories(true);
563
564    Router::new()
565        .route("/auth/exchange", post(auth_exchange))
566        .route("/v1/{*rest}", any(proxy_v1))
567        .route("/ws/events", any(ws_events))
568        // Fallback: serve the bundle (ServeDir). ServeDir already 405s on
569        // non-GET; we intercept earlier in the middleware chain to apply our
570        // own canonical Allow: GET header.
571        .fallback_service(serve_dir)
572        .layer(axum::middleware::from_fn(reject_non_get))
573        .with_state(state)
574}
575
576/// Middleware: every method except `GET` (and the special POST to
577/// `/auth/exchange`) returns 405 with `Allow: GET`. This is the structural
578/// enforcement of ADR-0016's read-only browser boundary.
579async fn reject_non_get(req: axum::http::Request<Body>, next: axum::middleware::Next) -> Response {
580    let method = req.method().clone();
581    let path = req.uri().path().to_string();
582
583    // /auth/exchange is the one exception — the bundle must POST it once to
584    // swap the fragment token for a cookie. WebSocket upgrade requests come
585    // through as GET, so that path is fine.
586    let is_auth_exchange = method == Method::POST && path == "/auth/exchange";
587    if method != Method::GET && !is_auth_exchange {
588        return method_not_allowed();
589    }
590    next.run(req).await
591}
592
593fn method_not_allowed() -> Response {
594    let mut resp = (StatusCode::METHOD_NOT_ALLOWED, "method not allowed\n").into_response();
595    resp.headers_mut()
596        .insert(header::ALLOW, HeaderValue::from_static("GET"));
597    resp
598}
599
600/// Cookie lifetime (red-team pass B, MED-1). Long enough to survive any
601/// realistic operator session, short enough that a stale cookie in a
602/// "restore tabs" browser is GC'd before the next `cellctl webui` run.
603const COOKIE_MAX_AGE_SECS: u64 = 86_400;
604
605/// `POST /auth/exchange` — body is `{"sess": "<base64-token>"}`. On match
606/// against the in-process session token, mint a cookie value and set
607/// `Set-Cookie: cellctl_session=<value>; HttpOnly; SameSite=Strict; Path=/;
608/// Max-Age=86400`.
609///
610/// Security gates (red-team pass B):
611///
612///   - CRIT-1: SINGLE USE. After the first successful call we atomically
613///     flip `exchange_consumed`; every subsequent call returns 401,
614///     regardless of whether the supplied token matches. This kills the
615///     race in which an attacker who observed the URL fragment could
616///     rotate the active cookie value out from under the operator.
617///   - HIGH-1: constant-time token comparison via `subtle::ConstantTimeEq`
618///     — Rust's `&` over `bool` is not guaranteed constant-time by the
619///     optimizer; the dedicated crate is.
620///   - MED-1: `Max-Age=86400` ensures browsers GC stale cookies after a
621///     day even on "restore tabs"-style long-lived sessions.
622///   - MED-2: require `Content-Type: application/json` (415 otherwise) so
623///     a future-loosened CORS posture can't enable a CSRF `<form>` post.
624async fn auth_exchange(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
625    use std::sync::atomic::Ordering;
626    use subtle::ConstantTimeEq;
627
628    // MED-2 — strict Content-Type gate. The bundle's fetch always sends
629    // application/json; anything else is either a misconfigured client or
630    // a CSRF attempt.
631    let ct_ok = headers
632        .get(header::CONTENT_TYPE)
633        .and_then(|v| v.to_str().ok())
634        .map(|s| {
635            // accept `application/json` or `application/json; charset=...`
636            let trimmed = s.split(';').next().unwrap_or("").trim();
637            trimmed.eq_ignore_ascii_case("application/json")
638        })
639        .unwrap_or(false);
640    if !ct_ok {
641        let mut resp = (
642            StatusCode::UNSUPPORTED_MEDIA_TYPE,
643            "Content-Type must be application/json\n",
644        )
645            .into_response();
646        resp.headers_mut().insert(
647            header::CONTENT_TYPE,
648            HeaderValue::from_static("text/plain; charset=utf-8"),
649        );
650        return resp;
651    }
652
653    // Parse the body explicitly so the Content-Type gate above runs even
654    // on malformed JSON (otherwise `Json<T>` parses before the gate).
655    let parsed: ExchangeRequest = match serde_json::from_slice(&body) {
656        Ok(p) => p,
657        Err(_) => return (StatusCode::BAD_REQUEST, "invalid json body\n").into_response(),
658    };
659
660    // CRIT-1 — single-use gate, checked BEFORE the constant-time compare
661    // so we don't even leak token-shape via timing on replays.
662    if state.exchange_consumed.load(Ordering::SeqCst) {
663        return (
664            StatusCode::UNAUTHORIZED,
665            "session token already exchanged\n",
666        )
667            .into_response();
668    }
669
670    // HIGH-1 — constant-time compare. Token is fixed-length (44 chars of
671    // URL_SAFE_NO_PAD over 32 bytes); even so, `subtle::ConstantTimeEq`
672    // protects against optimizer rewrites and any future change to a
673    // variable-length token format.
674    let provided = parsed.sess.as_bytes();
675    let expected = state.session_token.as_bytes();
676    if provided.len() != expected.len() || provided.ct_eq(expected).unwrap_u8() == 0 {
677        return (StatusCode::UNAUTHORIZED, "bad session token\n").into_response();
678    }
679
680    // Flip the consumed flag BEFORE minting the cookie. If two concurrent
681    // requests both passed the check above, only one will see
682    // `compare_exchange` return Ok — the other gets 401, preserving
683    // single-use semantics under load.
684    if state
685        .exchange_consumed
686        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
687        .is_err()
688    {
689        return (
690            StatusCode::UNAUTHORIZED,
691            "session token already exchanged\n",
692        )
693            .into_response();
694    }
695
696    // Mint the cookie value. Distinct from the URL token so we never reuse
697    // a value that might have appeared anywhere in history. rand 0.10:
698    // `random()` over the thread-local CSPRNG (see `mint_session_token`).
699    let buf: [u8; 32] = rand::random();
700    let cookie_value = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf);
701
702    {
703        let mut slot = state.session_cookie.write().await;
704        *slot = Some(cookie_value.clone());
705    }
706
707    // MED-1: include Max-Age so browsers GC the cookie even when the
708    // process is long gone (tab restore scenario).
709    let cookie_header = format!(
710        "cellctl_session={}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}",
711        cookie_value, COOKIE_MAX_AGE_SECS
712    );
713
714    let mut resp = (StatusCode::OK, "ok\n").into_response();
715    resp.headers_mut().insert(
716        header::SET_COOKIE,
717        HeaderValue::from_str(&cookie_header).unwrap(),
718    );
719    resp
720}
721
722/// Verify the request carries the `cellctl_session` cookie we set in
723/// `/auth/exchange`. Returns Some(()) if OK, None if missing/mismatch.
724async fn require_session_cookie(state: &AppState, headers: &HeaderMap) -> bool {
725    let expected = match state.session_cookie.read().await.clone() {
726        Some(v) => v,
727        None => return false,
728    };
729    let Some(cookie_hdr) = headers.get(header::COOKIE) else {
730        return false;
731    };
732    let Ok(cookie_str) = cookie_hdr.to_str() else {
733        return false;
734    };
735    // Cookies are `name=value; name=value` — find our entry.
736    for entry in cookie_str.split(';') {
737        let entry = entry.trim();
738        if let Some(v) = entry.strip_prefix("cellctl_session=") {
739            return v == expected;
740        }
741    }
742    false
743}
744
745/// `GET /v1/*` — reverse-proxy to upstream with Bearer injection.
746async fn proxy_v1(
747    State(state): State<AppState>,
748    AxumPath(rest): AxumPath<String>,
749    uri: Uri,
750    headers: HeaderMap,
751) -> Response {
752    if !require_session_cookie(&state, &headers).await {
753        return unauthorized();
754    }
755
756    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
757    let upstream_url = format!("{}/v1/{}{}", state.upstream_base, rest, query);
758
759    let client = match reqwest::Client::builder().build() {
760        Ok(c) => c,
761        Err(e) => return upstream_error(format!("client: {e}")),
762    };
763
764    let mut req = client.get(&upstream_url);
765    if let Some(tok) = state.upstream_bearer.as_ref() {
766        req = req.header(reqwest::header::AUTHORIZATION, format!("Bearer {tok}"));
767    }
768
769    let upstream_resp = match req.send().await {
770        Ok(r) => r,
771        Err(e) => return upstream_error(format!("send: {e}")),
772    };
773
774    let status = upstream_resp.status();
775    let resp_headers = upstream_resp.headers().clone();
776    let body_bytes = match upstream_resp.bytes().await {
777        Ok(b) => b,
778        Err(e) => return upstream_error(format!("read body: {e}")),
779    };
780
781    let mut out = Response::new(Body::from(body_bytes));
782    *out.status_mut() = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
783    // Forward content-type only (avoid leaking upstream Set-Cookie etc.).
784    if let Some(ct) = resp_headers.get(reqwest::header::CONTENT_TYPE) {
785        if let Ok(v) = HeaderValue::from_bytes(ct.as_bytes()) {
786            out.headers_mut().insert(header::CONTENT_TYPE, v);
787        }
788    }
789    out
790}
791
792/// `GET /ws/events` — WebSocket upgrade, forwarded to upstream.
793///
794/// axum 0.8 removed the blanket `Option<T>` extractor impl for plain
795/// `FromRequestParts` types — `Option<E>` now requires `E:
796/// OptionalFromRequestParts`, which `WebSocketUpgrade` does not implement.
797/// We therefore extract `WebSocketUpgrade` directly; axum rejects a
798/// non-upgrade request to this route before the handler runs. The
799/// session-cookie gate below still executes *before* `on_upgrade`, so an
800/// unauthenticated WebSocket client is rejected with 401 prior to any
801/// handshake completing.
802async fn ws_events(
803    State(state): State<AppState>,
804    uri: Uri,
805    headers: HeaderMap,
806    ws: WebSocketUpgrade,
807) -> Response {
808    if !require_session_cookie(&state, &headers).await {
809        return unauthorized();
810    }
811
812    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
813    let ws_url = {
814        // http:// → ws://, https:// → wss://
815        let base = state.upstream_base.as_str();
816        let ws_base = if let Some(rest) = base.strip_prefix("https://") {
817            format!("wss://{rest}")
818        } else if let Some(rest) = base.strip_prefix("http://") {
819            format!("ws://{rest}")
820        } else {
821            base.to_string()
822        };
823        format!("{ws_base}/ws/events{query}")
824    };
825    let bearer = state.upstream_bearer.as_ref().clone();
826    // Wave 2 red-team (MED-W2D-4): forward the client's
827    // `Sec-WebSocket-Protocol` to the upstream. axum's
828    // `WebSocketUpgrade::on_upgrade` finishes the handshake with the
829    // *client* without telling us which subprotocol(s) were requested, so
830    // we extract the header from the original request and propagate it
831    // verbatim. The upstream's choice (if any) is returned in its 101
832    // response — `connect_async` drops the response handle on the floor
833    // here, which matches axum's behaviour of accepting whichever
834    // subprotocol the underlying handler agreed to. If/when we want
835    // strict subprotocol negotiation, the upstream's `Sec-WebSocket-Protocol`
836    // header in `_response` would need to be plumbed back into the client
837    // accept frame; today's contract is "best-effort forwarding".
838    let subprotocols: Option<String> = headers
839        .get(axum::http::header::SEC_WEBSOCKET_PROTOCOL)
840        .and_then(|v| v.to_str().ok())
841        .map(|s| s.to_string());
842
843    ws.on_upgrade(move |client_ws| async move {
844        let (mut client_tx, mut client_rx) = client_ws.split();
845
846        // Build a request to upstream with the bearer header, since
847        // `connect_async` accepts a raw URL but loses our custom headers.
848        let mut request =
849            match tokio_tungstenite::tungstenite::client::IntoClientRequest::into_client_request(
850                ws_url.as_str(),
851            ) {
852                Ok(r) => r,
853                Err(_) => return,
854            };
855        if let Some(tok) = bearer {
856            if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(&format!(
857                "Bearer {tok}"
858            )) {
859                request.headers_mut().insert(
860                    tokio_tungstenite::tungstenite::http::header::AUTHORIZATION,
861                    v,
862                );
863            }
864        }
865        if let Some(proto) = subprotocols.as_ref() {
866            if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(proto) {
867                request.headers_mut().insert(
868                    tokio_tungstenite::tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL,
869                    v,
870                );
871            }
872        }
873        let (upstream_ws, _) = match tokio_tungstenite::connect_async(request).await {
874            Ok(p) => p,
875            Err(_) => return,
876        };
877        let (mut up_tx, mut up_rx) = upstream_ws.split();
878
879        loop {
880            tokio::select! {
881                msg = client_rx.next() => match msg {
882                    Some(Ok(axum::extract::ws::Message::Text(t))) => {
883                        // axum 0.8 and tungstenite 0.29 each carry their own
884                        // `Utf8Bytes` newtype; bridge them through `&str`
885                        // (zero extra UTF-8 validation, the bytes are already
886                        // known-valid on both sides).
887                        let _ = up_tx
888                            .send(tokio_tungstenite::tungstenite::Message::Text(
889                                t.as_str().into(),
890                            ))
891                            .await;
892                    }
893                    Some(Ok(axum::extract::ws::Message::Binary(b))) => {
894                        let _ = up_tx
895                            .send(tokio_tungstenite::tungstenite::Message::Binary(b))
896                            .await;
897                    }
898                    Some(Ok(axum::extract::ws::Message::Ping(p))) => {
899                        let _ = up_tx
900                            .send(tokio_tungstenite::tungstenite::Message::Ping(p))
901                            .await;
902                    }
903                    Some(Ok(axum::extract::ws::Message::Pong(p))) => {
904                        let _ = up_tx
905                            .send(tokio_tungstenite::tungstenite::Message::Pong(p))
906                            .await;
907                    }
908                    Some(Ok(axum::extract::ws::Message::Close(_))) | None => break,
909                    Some(Err(_)) => break,
910                },
911                msg = up_rx.next() => match msg {
912                    Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t))) => {
913                        let _ = client_tx
914                            .send(axum::extract::ws::Message::Text(t.as_str().into()))
915                            .await;
916                    }
917                    Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(b))) => {
918                        let _ = client_tx
919                            .send(axum::extract::ws::Message::Binary(b))
920                            .await;
921                    }
922                    Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(p))) => {
923                        let _ = client_tx
924                            .send(axum::extract::ws::Message::Ping(p))
925                            .await;
926                    }
927                    Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(p))) => {
928                        let _ = client_tx
929                            .send(axum::extract::ws::Message::Pong(p))
930                            .await;
931                    }
932                    Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => break,
933                    Some(Ok(tokio_tungstenite::tungstenite::Message::Frame(_))) => {}
934                    Some(Err(_)) => break,
935                },
936            }
937        }
938    })
939}
940
941fn unauthorized() -> Response {
942    (StatusCode::UNAUTHORIZED, "missing session cookie\n").into_response()
943}
944
945fn upstream_error(msg: String) -> Response {
946    (StatusCode::BAD_GATEWAY, format!("upstream: {msg}\n")).into_response()
947}
948
949// LOW-1 / LOW-2 (red-team pass B): the dead `state_upstream_for_log` and
950// `_BYTES_KEEP` shim are gone. `Bytes` is now used directly by
951// `auth_exchange` (CRIT-1 / MED-2 rewrite) and the startup banner reads
952// `upstream_log` directly in `run()`.
953
954#[allow(dead_code)]
955const _HEADER_NAME_KEEP: fn() = || {
956    // `HeaderName` is imported for future use in custom header-injection
957    // paths; keep this shim until that lands.
958    let _: HeaderName = HeaderName::from_static("x-keep");
959};
960
961// ---------------------------------------------------------------------------
962// Tests
963// ---------------------------------------------------------------------------
964
965#[cfg(test)]
966mod tests {
967    use super::*;
968    use axum::body::to_bytes;
969    use axum::http::Request;
970    use tower::ServiceExt; // for `oneshot`
971
972    fn test_state(bundle_dir: PathBuf) -> AppState {
973        AppState {
974            upstream_base: Arc::new("http://127.0.0.1:0".to_string()),
975            upstream_bearer: Arc::new(None),
976            session_token: Arc::new("test-token".to_string()),
977            session_cookie: Arc::new(RwLock::new(None)),
978            exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
979            bundle_dir: Arc::new(bundle_dir),
980        }
981    }
982
983    /// Use the real bundle if it's been built; otherwise create a fake one
984    /// in a tempdir with a minimal index.html.
985    fn ensure_bundle_dir() -> PathBuf {
986        let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
987        let real = manifest_dir.join("static");
988        if real.join("index.html").exists() {
989            return real;
990        }
991        // Fall back to a freshly-minted temp dir.
992        let tmp = std::env::temp_dir().join(format!("cellctl-webui-test-{}", std::process::id()));
993        std::fs::create_dir_all(&tmp).expect("mkdir tmp bundle");
994        std::fs::write(
995            tmp.join("index.html"),
996            "<!doctype html><title>cellctl webui</title>",
997        )
998        .expect("write index.html");
999        tmp
1000    }
1001
1002    #[tokio::test]
1003    async fn serves_index_at_root() {
1004        let bundle = ensure_bundle_dir();
1005        let app = build_router(test_state(bundle));
1006
1007        let resp = app
1008            .oneshot(
1009                Request::builder()
1010                    .method("GET")
1011                    .uri("/")
1012                    .body(Body::empty())
1013                    .unwrap(),
1014            )
1015            .await
1016            .unwrap();
1017        assert_eq!(resp.status(), StatusCode::OK);
1018        let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
1019        let body_str = std::str::from_utf8(&body).unwrap();
1020        assert!(
1021            body_str.to_ascii_lowercase().contains("<!doctype html")
1022                || body_str.to_ascii_lowercase().contains("<html"),
1023            "expected HTML at /, got: {body_str:.200}"
1024        );
1025    }
1026
1027    #[tokio::test]
1028    async fn non_get_returns_405() {
1029        let bundle = ensure_bundle_dir();
1030        let app = build_router(test_state(bundle));
1031
1032        let resp = app
1033            .oneshot(
1034                Request::builder()
1035                    .method("DELETE")
1036                    .uri("/v1/formations")
1037                    .body(Body::empty())
1038                    .unwrap(),
1039            )
1040            .await
1041            .unwrap();
1042        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
1043        assert_eq!(
1044            resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
1045            Some(b"GET" as &[u8]),
1046        );
1047    }
1048
1049    #[tokio::test]
1050    async fn put_to_v1_returns_405() {
1051        let bundle = ensure_bundle_dir();
1052        let app = build_router(test_state(bundle));
1053
1054        let resp = app
1055            .oneshot(
1056                Request::builder()
1057                    .method("PUT")
1058                    .uri("/v1/formations/foo")
1059                    .body(Body::from("body"))
1060                    .unwrap(),
1061            )
1062            .await
1063            .unwrap();
1064        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
1065    }
1066
1067    #[tokio::test]
1068    async fn proxy_without_cookie_returns_401() {
1069        let bundle = ensure_bundle_dir();
1070        let app = build_router(test_state(bundle));
1071
1072        let resp = app
1073            .oneshot(
1074                Request::builder()
1075                    .method("GET")
1076                    .uri("/v1/formations")
1077                    .body(Body::empty())
1078                    .unwrap(),
1079            )
1080            .await
1081            .unwrap();
1082        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1083    }
1084
1085    #[tokio::test]
1086    async fn auth_exchange_with_wrong_token_returns_401() {
1087        let bundle = ensure_bundle_dir();
1088        let app = build_router(test_state(bundle));
1089
1090        let resp = app
1091            .oneshot(
1092                Request::builder()
1093                    .method("POST")
1094                    .uri("/auth/exchange")
1095                    .header(header::CONTENT_TYPE, "application/json")
1096                    .body(Body::from(r#"{"sess":"wrong"}"#))
1097                    .unwrap(),
1098            )
1099            .await
1100            .unwrap();
1101        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1102    }
1103
1104    #[tokio::test]
1105    async fn auth_exchange_with_right_token_sets_cookie() {
1106        let bundle = ensure_bundle_dir();
1107        let state = test_state(bundle);
1108        let app = build_router(state.clone());
1109
1110        let resp = app
1111            .oneshot(
1112                Request::builder()
1113                    .method("POST")
1114                    .uri("/auth/exchange")
1115                    .header(header::CONTENT_TYPE, "application/json")
1116                    .body(Body::from(r#"{"sess":"test-token"}"#))
1117                    .unwrap(),
1118            )
1119            .await
1120            .unwrap();
1121        assert_eq!(resp.status(), StatusCode::OK);
1122        let cookie = resp
1123            .headers()
1124            .get(header::SET_COOKIE)
1125            .expect("Set-Cookie header present")
1126            .to_str()
1127            .unwrap()
1128            .to_string();
1129        assert!(cookie.starts_with("cellctl_session="));
1130        assert!(cookie.contains("HttpOnly"));
1131        assert!(cookie.contains("SameSite=Strict"));
1132
1133        // And the state should now hold a matching cookie value.
1134        let stored = state.session_cookie.read().await.clone();
1135        assert!(stored.is_some());
1136        let stored = stored.unwrap();
1137        assert!(cookie.contains(&format!("cellctl_session={stored}")));
1138    }
1139
1140    #[tokio::test]
1141    async fn proxy_with_valid_cookie_attempts_upstream() {
1142        // After /auth/exchange succeeds, a subsequent GET /v1/* should *not*
1143        // 401 — it'll instead 502 (bad gateway) because there's no real
1144        // upstream listening, which is exactly what proves the cookie gate
1145        // passed.
1146        let bundle = ensure_bundle_dir();
1147        let state = test_state(bundle);
1148        let app = build_router(state.clone());
1149
1150        let exch = app
1151            .clone()
1152            .oneshot(
1153                Request::builder()
1154                    .method("POST")
1155                    .uri("/auth/exchange")
1156                    .header(header::CONTENT_TYPE, "application/json")
1157                    .body(Body::from(r#"{"sess":"test-token"}"#))
1158                    .unwrap(),
1159            )
1160            .await
1161            .unwrap();
1162        assert_eq!(exch.status(), StatusCode::OK);
1163        let cookie_hdr = exch
1164            .headers()
1165            .get(header::SET_COOKIE)
1166            .unwrap()
1167            .to_str()
1168            .unwrap()
1169            .to_string();
1170        // Strip attributes; keep just `cellctl_session=<value>`.
1171        let cookie_pair = cookie_hdr.split(';').next().unwrap().trim().to_string();
1172
1173        let resp = app
1174            .oneshot(
1175                Request::builder()
1176                    .method("GET")
1177                    .uri("/v1/formations")
1178                    .header(header::COOKIE, cookie_pair)
1179                    .body(Body::empty())
1180                    .unwrap(),
1181            )
1182            .await
1183            .unwrap();
1184        // No live upstream → bad gateway, not 401.
1185        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
1186    }
1187
1188    /// Proxy invariant: no matter what HTTP method the browser sends, the
1189    /// upstream cellos-server must NEVER see a non-GET. The proxy is the
1190    /// structural enforcement of ADR-0016's read-only browser boundary
1191    /// (ADR-0017 §Decision 4, item 5: "Refuse non-GET methods with HTTP 405").
1192    ///
1193    /// We spin up a tiny inline mock upstream on a real loopback port that
1194    /// counts the methods it sees, point AppState at it, and fire a battery of
1195    /// inbound methods through the proxy. The assertion is bidirectional:
1196    ///
1197    ///   1. The mock saw ZERO non-GET requests.
1198    ///   2. Every non-GET inbound request returned 405 with `Allow: GET`.
1199    #[tokio::test]
1200    async fn proxy_only_emits_get_to_upstream() {
1201        use std::sync::Mutex;
1202        use tokio::io::{AsyncReadExt, AsyncWriteExt};
1203        use tokio::net::TcpListener;
1204
1205        // ---- inline mock upstream ----
1206        let mock = TcpListener::bind("127.0.0.1:0").await.expect("bind mock");
1207        let mock_addr = mock.local_addr().expect("mock addr");
1208        let methods_seen: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
1209        let methods_for_task = methods_seen.clone();
1210
1211        // Spawn a tiny accept loop that reads the request line, records the
1212        // method, then writes a minimal `HTTP/1.1 200 OK` and closes. We do
1213        // not implement keep-alive, content-length parsing, or anything else
1214        // — the proxy doesn't keep connections open across requests against
1215        // this mock anyway (each `reqwest::Client::builder().build()` builds a
1216        // fresh client).
1217        tokio::spawn(async move {
1218            loop {
1219                let (mut stream, _) = match mock.accept().await {
1220                    Ok(p) => p,
1221                    Err(_) => return,
1222                };
1223                let methods = methods_for_task.clone();
1224                tokio::spawn(async move {
1225                    let mut buf = [0u8; 4096];
1226                    let n = match stream.read(&mut buf).await {
1227                        Ok(n) => n,
1228                        Err(_) => return,
1229                    };
1230                    let head = String::from_utf8_lossy(&buf[..n]).to_string();
1231                    if let Some(first_line) = head.lines().next() {
1232                        if let Some(method) = first_line.split_whitespace().next() {
1233                            methods.lock().unwrap().push(method.to_string());
1234                        }
1235                    }
1236                    let _ = stream
1237                        .write_all(
1238                            b"HTTP/1.1 200 OK\r\n\
1239                              Content-Type: application/json\r\n\
1240                              Content-Length: 2\r\n\
1241                              Connection: close\r\n\
1242                              \r\n\
1243                              {}",
1244                        )
1245                        .await;
1246                    let _ = stream.shutdown().await;
1247                });
1248            }
1249        });
1250
1251        // ---- proxy under test ----
1252        let bundle = ensure_bundle_dir();
1253        let state = AppState {
1254            upstream_base: Arc::new(format!("http://{}", mock_addr)),
1255            upstream_bearer: Arc::new(Some("test-bearer".to_string())),
1256            session_token: Arc::new("test-token".to_string()),
1257            session_cookie: Arc::new(RwLock::new(None)),
1258            exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
1259            bundle_dir: Arc::new(bundle),
1260        };
1261        let app = build_router(state.clone());
1262
1263        // Mint a session cookie so /v1/* GETs can pass the auth gate. (Non-GET
1264        // requests will be rejected at the 405 middleware *before* the cookie
1265        // check, so this only matters for the GET-passes-through assertion.)
1266        let exch = app
1267            .clone()
1268            .oneshot(
1269                Request::builder()
1270                    .method("POST")
1271                    .uri("/auth/exchange")
1272                    .header(header::CONTENT_TYPE, "application/json")
1273                    .body(Body::from(r#"{"sess":"test-token"}"#))
1274                    .unwrap(),
1275            )
1276            .await
1277            .unwrap();
1278        assert_eq!(exch.status(), StatusCode::OK);
1279        let cookie_pair = exch
1280            .headers()
1281            .get(header::SET_COOKIE)
1282            .unwrap()
1283            .to_str()
1284            .unwrap()
1285            .split(';')
1286            .next()
1287            .unwrap()
1288            .trim()
1289            .to_string();
1290
1291        // Fire a battery of inbound methods. Each non-GET MUST return 405 and
1292        // MUST NOT touch the upstream.
1293        let non_get_methods = ["POST", "PUT", "DELETE", "PATCH"];
1294        for m in non_get_methods {
1295            let resp = app
1296                .clone()
1297                .oneshot(
1298                    Request::builder()
1299                        .method(m)
1300                        .uri("/v1/formations")
1301                        .header(header::COOKIE, cookie_pair.clone())
1302                        .body(Body::from("payload that must never reach upstream"))
1303                        .unwrap(),
1304                )
1305                .await
1306                .unwrap();
1307            assert_eq!(
1308                resp.status(),
1309                StatusCode::METHOD_NOT_ALLOWED,
1310                "inbound {m} should be 405"
1311            );
1312            assert_eq!(
1313                resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
1314                Some(b"GET" as &[u8]),
1315                "405 response for {m} must carry `Allow: GET`"
1316            );
1317        }
1318
1319        // Now fire an actual GET, which SHOULD reach the mock upstream.
1320        let get_resp = app
1321            .clone()
1322            .oneshot(
1323                Request::builder()
1324                    .method("GET")
1325                    .uri("/v1/formations")
1326                    .header(header::COOKIE, cookie_pair.clone())
1327                    .body(Body::empty())
1328                    .unwrap(),
1329            )
1330            .await
1331            .unwrap();
1332        // The GET must NOT be rejected by the proxy's read-only gate.
1333        assert_ne!(
1334            get_resp.status(),
1335            StatusCode::METHOD_NOT_ALLOWED,
1336            "GET must pass the 405 middleware"
1337        );
1338
1339        // Give the mock a moment to record the inbound connection.
1340        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1341
1342        // ---- assertions on the mock's observation log ----
1343        let observed = methods_seen.lock().unwrap().clone();
1344        // Exactly one upstream request (the GET) — the four non-GETs were
1345        // all stopped at the proxy.
1346        let non_get_count = observed.iter().filter(|m| m.as_str() != "GET").count();
1347        assert_eq!(
1348            non_get_count, 0,
1349            "upstream saw non-GET method(s): {observed:?}"
1350        );
1351        assert!(
1352            observed.iter().any(|m| m == "GET"),
1353            "expected at least one GET to reach upstream, saw {observed:?}"
1354        );
1355    }
1356}