cellos_ctl/cmd/

webui.rs

//! `cellctl webui` — localhost reverse proxy + static bundle host.
//!
//! Per ADR-0017, the web view ships *with cellctl*, not with cellos-server.
//! Operators invoke `cellctl webui` to spin up a foreground localhost proxy:
//!
//!   - Serves `crates/cellos-ctl/static/` (the Vite build output) at `/`.
//!   - Reverse-proxies `GET /v1/*` and `GET /ws/events` upstream to the
//!     cellos-server URL configured in `~/.cellctl/config`, injecting the
//!     bearer token on the way out. The bundle never sees the token.
//!   - Refuses any non-`GET` method with HTTP 405 (`Allow: GET`). This is the
//!     structural enforcement of ADR-0016's read-only browser boundary.
//!   - Binds a session token in a URL fragment (`/#sess=<base64>`) and
//!     swaps it for an `HttpOnly; SameSite=Strict` cookie via
//!     `POST /auth/exchange`. Subsequent proxy + WS requests require the
//!     cookie. The fragment is cleared by the bundle's bootstrap once
//!     the cookie is set.
//!   - Exits cleanly on SIGINT.
//!
//! Bind modes (ADR-0017 §Decision 4):
//!
//! - `--bind auto` (default, on Unix): bind BOTH a loopback TCP port (for the
//!   browser) AND a Unix socket at
//!   `${XDG_RUNTIME_DIR:-/tmp}/cellctl-webui-<pid>.sock` (for inter-process
//!   tooling that wants to bypass loopback). On Windows, `auto` degrades to
//!   loopback-only.
//! - `--bind loopback`: TCP loopback only.
//! - `--bind unix` (Unix only): Unix socket only. The browser cannot reach a
//!   Unix socket directly — this mode is for inter-process forwarders (e.g.
//!   `socat` / `ssh -L`). On Windows this errors out.
//!
//! The Unix socket is created with mode `0600` (operator-owned only) and is
//! removed on graceful shutdown (SIGINT).

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use axum::body::{Body, Bytes};
use axum::extract::{ws::WebSocketUpgrade, Path as AxumPath, State};
use axum::http::{header, HeaderMap, HeaderName, HeaderValue, Method, StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, post};
use axum::Router;
use base64::Engine as _;
use clap::ValueEnum;
use futures_util::{SinkExt, StreamExt};
use rand::RngCore;
use serde::Deserialize;
use tokio::sync::RwLock;
use tower_http::services::ServeDir;

use crate::client::CellosClient;
use crate::config::Config;
use crate::exit::{CtlError, CtlResult};

/// Where the bundle lives on disk. Resolved relative to the cellctl binary's
/// crate at compile time; we look up the actual path at runtime via the
/// `CARGO_MANIFEST_DIR` env var (set in tests) or `./static` (when the
/// binary is installed alongside its bundle).
const BUNDLE_DIR_RELATIVE: &str = "static";

/// Bind mode for the webui proxy.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, ValueEnum)]
pub enum BindMode {
    /// Default: on Unix, bind BOTH a loopback TCP port and a Unix socket.
    /// On Windows, degrade to loopback-only.
    #[default]
    Auto,
    /// Force 127.0.0.1 on a random high port. No Unix socket.
    Loopback,
    /// Force a Unix socket only (no TCP). Browsers cannot reach a Unix
    /// socket directly; this mode is for inter-process forwarders. Errors
    /// on Windows.
    Unix,
}

/// Per-process shared state: upstream client + the one valid session token
/// (until it's exchanged) and the resulting session cookie value.
///
/// SECURITY NOTE (red-team pass B, HIGH-5): never log `upstream_bearer` or
/// `session_token` outside this module. The proxy injects `Authorization:
/// Bearer ...` headers into outbound requests; do NOT enable
/// `RUST_LOG=reqwest=trace` in production — `reqwest` will dump those
/// headers to stderr. The audit log path in `cmd/webui.rs` keeps the
/// startup banner deliberately bearer-free.
#[derive(Clone)]
struct AppState {
    /// Upstream cellos-server base URL (e.g. `http://127.0.0.1:8080`).
    upstream_base: Arc<String>,
    /// Upstream Bearer token to inject on every proxied request, if any.
    upstream_bearer: Arc<Option<String>>,
    /// The unguessable, single-use session token printed in the URL fragment.
    session_token: Arc<String>,
    /// Once `/auth/exchange` succeeds, this holds the cookie value the
    /// browser is expected to present on every subsequent request.
    session_cookie: Arc<RwLock<Option<String>>>,
    /// Single-use gate (red-team pass B, CRIT-1): once `/auth/exchange` has
    /// been successfully called, every subsequent call MUST return 401
    /// regardless of whether the provided token matches. Closes the
    /// "attacker steals the URL fragment, races the operator" path.
    exchange_consumed: Arc<std::sync::atomic::AtomicBool>,
    /// Filesystem path of the static bundle (for `ServeDir`).
    bundle_dir: Arc<PathBuf>,
}

#[derive(Deserialize)]
struct ExchangeRequest {
    sess: String,
}

/// Entry point invoked by `main.rs`.
pub async fn run(cfg: &Config, open: bool, bind: BindMode) -> CtlResult<()> {
    // Reuse the same effective config CellosClient sees, so the URL/token
    // injected into the proxy matches what the rest of cellctl would use.
    // Touching CellosClient also validates that the token (if any) parses as
    // a header value.
    let _ = CellosClient::new(cfg)?;

    let upstream_base = cfg.effective_server().trim_end_matches('/').to_string();
    let upstream_bearer = cfg.effective_token();

    let bundle_dir = resolve_bundle_dir()?;
    if !bundle_dir.join("index.html").exists() {
        return Err(CtlError::usage(format!(
            "webui bundle not found at {}/index.html — run `npm --prefix web run build` first",
            bundle_dir.display()
        )));
    }

    let session_token = mint_session_token();

    let upstream_log = upstream_base.clone();
    let state = AppState {
        upstream_base: Arc::new(upstream_base),
        upstream_bearer: Arc::new(upstream_bearer),
        session_token: Arc::new(session_token.clone()),
        session_cookie: Arc::new(RwLock::new(None)),
        exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        bundle_dir: Arc::new(bundle_dir.clone()),
    };

    let app = build_router(state);

    // Decide which listeners we actually want for this run.
    let (want_tcp, want_unix) = resolve_bind_plan(bind)?;

    let tcp_listener = if want_tcp {
        let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
        Some(
            tokio::net::TcpListener::bind(addr)
                .await
                .map_err(|e| CtlError::usage(format!("bind 127.0.0.1: {e}")))?,
        )
    } else {
        None
    };

    #[cfg(unix)]
    let unix_socket_path: Option<PathBuf> = if want_unix {
        Some(unix_socket_path_for_pid(std::process::id()))
    } else {
        None
    };
    #[cfg(not(unix))]
    let unix_socket_path: Option<PathBuf> = None;

    let browser_url = if let Some(l) = tcp_listener.as_ref() {
        let local_addr = l
            .local_addr()
            .map_err(|e| CtlError::usage(format!("local_addr: {e}")))?;
        Some(format!("http://{}/#sess={}", local_addr, session_token))
    } else {
        None
    };

    if let Some(url) = browser_url.as_ref() {
        println!("cellctl webui: {}", url);
    }
    if let Some(p) = unix_socket_path.as_ref() {
        println!("cellctl webui: unix://{}", p.display());
    }
    if browser_url.is_none() {
        // unix-only mode: be loud that the browser can't reach this.
        eprintln!(
            "cellctl webui: --bind unix has no browser-reachable URL; \
             use a forwarder (e.g. `socat TCP-LISTEN:0,reuseaddr,fork UNIX-CONNECT:{}`) \
             or rerun with `--bind auto` for a loopback URL.",
            unix_socket_path
                .as_ref()
                .map(|p| p.display().to_string())
                .unwrap_or_else(|| "<socket>".to_string())
        );
    }
    // LOW-1 (red-team pass B): print the actual upstream URL the proxy will
    // forward to, not the dead "(see state)" placeholder. The bearer token
    // is never printed — only the base URL.
    println!("upstream: {}", upstream_log);
    println!("press Ctrl-C to stop");

    if open {
        if let Some(url) = browser_url.as_ref() {
            // MED-4 (red-team pass B): belt-and-suspenders sanity check on
            // the URL before handing it to `opener` (which shells out to
            // `xdg-open` / `open` / `Start-Process`). Today the URL is built
            // from `local_addr` + a base64-url-no-pad token — both
            // controlled — but a future contributor might add a header
            // param that smuggles control characters. Reject anything that
            // isn't a clean loopback http URL.
            if !is_safe_open_url(url) {
                eprintln!(
                    "cellctl webui: refusing to open URL (failed loopback-http sanity check)"
                );
            } else if let Err(e) = opener::open(url) {
                eprintln!("cellctl webui: could not launch browser: {e}");
            }
        } else {
            eprintln!("cellctl webui: --open ignored: no loopback URL bound (use --bind auto)");
        }
    }

    // Build a shutdown signal future shared by both listeners. We resolve on
    // SIGINT (Ctrl-C) and SIGTERM (kubernetes / systemd / `kill <pid>`) and
    // (best-effort) clean up the socket file before either listener unbinds
    // it. HIGH-6 (red-team pass B): without the SIGTERM handler, container
    // orchestrators hard-kill the process and leave a stale Unix socket
    // file behind for the next run to trip over.
    let (shutdown_tx, _shutdown_rx) = tokio::sync::broadcast::channel::<()>(1);
    {
        let shutdown_tx = shutdown_tx.clone();
        tokio::spawn(async move {
            wait_for_shutdown_signal().await;
            eprintln!("shutting down");
            let _ = shutdown_tx.send(());
        });
    }

    // Spawn the TCP server if requested.
    let tcp_task = if let Some(listener) = tcp_listener {
        let app = app.clone();
        let mut rx = shutdown_tx.subscribe();
        Some(tokio::spawn(async move {
            axum::serve(listener, app)
                .with_graceful_shutdown(async move {
                    let _ = rx.recv().await;
                })
                .await
        }))
    } else {
        None
    };

    // Spawn the Unix server if requested.
    #[cfg(unix)]
    let unix_task = if let Some(path) = unix_socket_path.clone() {
        let app = app.clone();
        let mut rx = shutdown_tx.subscribe();
        let listener = bind_unix_listener(&path)?;
        Some(tokio::spawn(async move {
            serve_unix(listener, app, async move {
                let _ = rx.recv().await;
            })
            .await
        }))
    } else {
        None
    };
    #[cfg(not(unix))]
    let unix_task: Option<tokio::task::JoinHandle<std::io::Result<()>>> = None;

    // Wait for both servers (whichever exist). Capture the first error.
    let mut first_err: Option<String> = None;
    if let Some(t) = tcp_task {
        match t.await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                let _ = first_err.get_or_insert_with(|| format!("tcp: {e}"));
            }
            Err(e) => {
                let _ = first_err.get_or_insert_with(|| format!("tcp join: {e}"));
            }
        }
    }
    if let Some(t) = unix_task {
        match t.await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                let _ = first_err.get_or_insert_with(|| format!("unix: {e}"));
            }
            Err(e) => {
                let _ = first_err.get_or_insert_with(|| format!("unix join: {e}"));
            }
        }
    }

    // Best-effort cleanup: remove the socket file. (UnixListener does not unlink
    // on drop.)
    if let Some(p) = unix_socket_path.as_ref() {
        let _ = std::fs::remove_file(p);
    }

    if let Some(e) = first_err {
        return Err(CtlError::api(format!("webui server: {e}")));
    }
    Ok(())
}

/// Decide which listeners to spawn for a given bind mode.
///
/// Returns `(want_tcp, want_unix)`. On non-Unix platforms, `want_unix` is
/// always false and `BindMode::Unix` errors.
fn resolve_bind_plan(bind: BindMode) -> CtlResult<(bool, bool)> {
    #[cfg(unix)]
    {
        Ok(match bind {
            BindMode::Auto => (true, true),
            BindMode::Loopback => (true, false),
            BindMode::Unix => (false, true),
        })
    }
    #[cfg(not(unix))]
    {
        Ok(match bind {
            BindMode::Auto => (true, false),
            BindMode::Loopback => (true, false),
            BindMode::Unix => {
                return Err(CtlError::usage(
                    "--bind unix is not supported on Windows; use --bind loopback (the default)",
                ));
            }
        })
    }
}

/// Compute the Unix socket path for a given pid.
///
/// Preferred parent: `$XDG_RUNTIME_DIR` (operator-owned, mode 0700 on most
/// distros). Fallback: `/tmp/cellctl-webui-<uid>/` — a per-uid subdirectory
/// we create with mode 0700 so the world-writable sticky `/tmp` cannot host
/// the socket directly. HIGH-2 (red-team pass B): the previous "drop the
/// socket straight into /tmp" fallback was vulnerable to a same-host
/// attacker with write access to `/tmp` swapping symlinks under the chmod
/// call.
#[cfg(unix)]
fn unix_socket_path_for_pid(pid: u32) -> PathBuf {
    let dir = if let Some(xdg) = std::env::var_os("XDG_RUNTIME_DIR") {
        PathBuf::from(xdg)
    } else {
        // Fallback: per-uid sub-directory in /tmp so we never bind directly
        // under a world-writable parent.
        let uid = unsafe { libc::getuid() };
        PathBuf::from("/tmp").join(format!("cellctl-{uid}"))
    };
    dir.join(format!("cellctl-webui-{pid}.sock"))
}

/// Bind a Unix socket at `path` with mode 0600.
///
/// HIGH-2 (red-team pass B): the previous implementation called
/// `set_permissions(path, 0600)` AFTER `bind`, which follows symlinks — an
/// attacker who can write to the parent directory could symlink-swap the
/// socket and trick us into chmodding an attacker-chosen target. We close
/// the window two ways:
///
///   1. Ensure the parent directory exists with mode 0700 (operator-only).
///      If `$XDG_RUNTIME_DIR` is unset, we use a per-uid sub-directory of
///      `/tmp` that we create ourselves rather than binding directly into
///      world-writable `/tmp`.
///   2. Set `umask(0o077)` immediately before `bind` and restore it after.
///      `UnixListener::bind` creates the socket file with permissions
///      `0o666 & !umask` — with umask 0o077 we land on 0o600 atomically,
///      so there is no observable window where the socket exists with
///      broader permissions. The trailing `set_permissions` becomes a
///      belt-and-suspenders idempotent.
#[cfg(unix)]
fn bind_unix_listener(path: &std::path::Path) -> CtlResult<tokio::net::UnixListener> {
    use std::os::unix::fs::PermissionsExt;

    // Ensure parent directory exists and is 0700.
    if let Some(parent) = path.parent() {
        if !parent.exists() {
            std::fs::create_dir_all(parent)
                .map_err(|e| CtlError::usage(format!("mkdir {}: {e}", parent.display())))?;
        }
        // Tighten to 0700 unconditionally — if someone else owns the parent
        // with broader perms, set_permissions will return EPERM, which we
        // surface to the operator.
        let _ = std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700));
    }

    // Remove a stale socket file from a prior crashed run. Best-effort.
    let _ = std::fs::remove_file(path);

    // Atomic mode-0600 bind via tight umask. SAFETY: umask is a process-
    // global setting; we hold it tight only across the bind call and
    // restore it immediately. The restore is essential — without it, any
    // file the rest of the process creates inherits 0600, which surprises
    // future contributors and breaks tracing-subscriber log file output.
    let prev_umask = unsafe { libc::umask(0o077) };
    let bind_result = tokio::net::UnixListener::bind(path);
    unsafe {
        let _ = libc::umask(prev_umask);
    }
    let listener =
        bind_result.map_err(|e| CtlError::usage(format!("bind {}: {e}", path.display())))?;

    // Belt-and-suspenders: ensure final mode is 0600. NOTE: this still
    // follows symlinks, but the umask-driven atomic bind above already
    // produced the right mode; this call exists to catch the edge case
    // where the platform's bind() ignored umask for some reason.
    let perms = std::fs::Permissions::from_mode(0o600);
    std::fs::set_permissions(path, perms)
        .map_err(|e| CtlError::usage(format!("chmod 0600 {}: {e}", path.display())))?;

    Ok(listener)
}

/// HIGH-6 (red-team pass B): wait for either SIGINT (Ctrl-C) or, on Unix,
/// SIGTERM. On Windows, only Ctrl-C is observable; SIGTERM doesn't exist.
async fn wait_for_shutdown_signal() {
    #[cfg(unix)]
    {
        use tokio::signal::unix::{signal, SignalKind};
        let mut sigint = match signal(SignalKind::interrupt()) {
            Ok(s) => s,
            Err(_) => {
                let _ = tokio::signal::ctrl_c().await;
                return;
            }
        };
        let mut sigterm = match signal(SignalKind::terminate()) {
            Ok(s) => s,
            Err(_) => {
                let _ = sigint.recv().await;
                return;
            }
        };
        tokio::select! {
            _ = sigint.recv() => {},
            _ = sigterm.recv() => {},
        }
    }
    #[cfg(not(unix))]
    {
        let _ = tokio::signal::ctrl_c().await;
    }
}

/// MED-4 (red-team pass B): sanity check the URL we're about to hand to
/// `opener::open`. Belt-and-suspenders — today this is always a clean
/// loopback http URL, but a future contributor adding a query param could
/// smuggle control characters into a shell-out. Reject anything that:
///
///   - doesn't parse as a `url::Url`
///   - isn't `http` or `https`
///   - resolves to a non-loopback host
///   - contains any ASCII control character anywhere in the URL string
fn is_safe_open_url(s: &str) -> bool {
    // Any control character (incl. CR, LF, NUL, escape) is an immediate
    // reject — these can shell-escape on some platforms.
    if s.chars().any(|c| c.is_control()) {
        return false;
    }
    let Ok(u) = url::Url::parse(s) else {
        return false;
    };
    let scheme = u.scheme();
    if scheme != "http" && scheme != "https" {
        return false;
    }
    let Some(host) = u.host_str() else {
        return false;
    };
    // Only loopback. cellctl webui binds 127.0.0.1 — refuse if the URL
    // somehow points elsewhere.
    matches!(host, "127.0.0.1" | "localhost" | "::1" | "[::1]")
}

/// Serve `app` over a Unix-domain `UnixListener` with graceful shutdown.
///
/// `axum::serve` requires a `tokio::net::TcpListener`, so for Unix sockets we
/// drop down to the lower-level `accept_loop`-style approach using `hyper`
/// directly via `tower::Service::call`.
#[cfg(unix)]
async fn serve_unix(
    listener: tokio::net::UnixListener,
    app: Router,
    shutdown: impl std::future::Future<Output = ()>,
) -> std::io::Result<()> {
    use std::convert::Infallible;
    use tower::Service;

    // axum::Router implements tower::Service<Request<Body>>. We accept Unix
    // connections and hand each one to a hyper http1 connection driving the
    // router. This mirrors what axum::serve does internally for TCP, minus the
    // TCP-specific bits.
    tokio::pin!(shutdown);

    loop {
        tokio::select! {
            _ = &mut shutdown => return Ok(()),
            accepted = listener.accept() => {
                let (stream, _peer) = match accepted {
                    Ok(s) => s,
                    Err(e) => {
                        // EMFILE / transient — log and continue, do not crash.
                        eprintln!("webui: unix accept error: {e}");
                        continue;
                    }
                };
                let app = app.clone();
                tokio::spawn(async move {
                    // `service_fn` requires `Fn`, but Router::call needs
                    // `&mut self`. Clone the router per request so each
                    // invocation owns its own mutable handle.
                    let svc = hyper::service::service_fn(move |req: http::Request<hyper::body::Incoming>| {
                        let mut router = app.clone();
                        async move {
                            let resp: Response = match router.call(req.map(Body::new)).await {
                                Ok(r) => r,
                                Err(never) => match never {},
                            };
                            Ok::<_, Infallible>(resp)
                        }
                    });
                    let io = hyper_util::rt::TokioIo::new(stream);
                    let _ = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new())
                        .serve_connection_with_upgrades(io, svc)
                        .await;
                });
            }
        }
    }
}

/// Mint a 32-byte random session token and base64-url-encode it.
fn mint_session_token() -> String {
    let mut buf = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut buf);
    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf)
}

/// Resolve the bundle directory. Tests set `CELLCTL_WEBUI_BUNDLE_DIR`
/// explicitly. Installed binaries look next to `crates/cellos-ctl/static/`
/// via `CARGO_MANIFEST_DIR` (build-time embed) or the current working dir.
fn resolve_bundle_dir() -> CtlResult<PathBuf> {
    if let Ok(p) = std::env::var("CELLCTL_WEBUI_BUNDLE_DIR") {
        return Ok(PathBuf::from(p));
    }
    // At build time we record the crate root; installed binaries can still
    // override via the env var above.
    let from_manifest = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(BUNDLE_DIR_RELATIVE);
    Ok(from_manifest)
}

/// Build the axum router with the static fileserver, /auth/exchange handler,
/// and the catch-all proxy route.
fn build_router(state: AppState) -> Router {
    // Static bundle: ServeDir handles GET only — non-GET on the static tree
    // falls through to our catch-all 405 layer.
    let bundle_dir = state.bundle_dir.as_ref().clone();
    let serve_dir = ServeDir::new(&bundle_dir).append_index_html_on_directories(true);

    Router::new()
        .route("/auth/exchange", post(auth_exchange))
        .route("/v1/*rest", any(proxy_v1))
        .route("/ws/events", any(ws_events))
        // Fallback: serve the bundle (ServeDir). ServeDir already 405s on
        // non-GET; we intercept earlier in the middleware chain to apply our
        // own canonical Allow: GET header.
        .fallback_service(serve_dir)
        .layer(axum::middleware::from_fn(reject_non_get))
        .with_state(state)
}

/// Middleware: every method except `GET` (and the special POST to
/// `/auth/exchange`) returns 405 with `Allow: GET`. This is the structural
/// enforcement of ADR-0016's read-only browser boundary.
async fn reject_non_get(req: axum::http::Request<Body>, next: axum::middleware::Next) -> Response {
    let method = req.method().clone();
    let path = req.uri().path().to_string();

    // /auth/exchange is the one exception — the bundle must POST it once to
    // swap the fragment token for a cookie. WebSocket upgrade requests come
    // through as GET, so that path is fine.
    let is_auth_exchange = method == Method::POST && path == "/auth/exchange";
    if method != Method::GET && !is_auth_exchange {
        return method_not_allowed();
    }
    next.run(req).await
}

fn method_not_allowed() -> Response {
    let mut resp = (StatusCode::METHOD_NOT_ALLOWED, "method not allowed\n").into_response();
    resp.headers_mut()
        .insert(header::ALLOW, HeaderValue::from_static("GET"));
    resp
}

/// Cookie lifetime (red-team pass B, MED-1). Long enough to survive any
/// realistic operator session, short enough that a stale cookie in a
/// "restore tabs" browser is GC'd before the next `cellctl webui` run.
const COOKIE_MAX_AGE_SECS: u64 = 86_400;

/// `POST /auth/exchange` — body is `{"sess": "<base64-token>"}`. On match
/// against the in-process session token, mint a cookie value and set
/// `Set-Cookie: cellctl_session=<value>; HttpOnly; SameSite=Strict; Path=/;
/// Max-Age=86400`.
///
/// Security gates (red-team pass B):
///
///   - CRIT-1: SINGLE USE. After the first successful call we atomically
///     flip `exchange_consumed`; every subsequent call returns 401,
///     regardless of whether the supplied token matches. This kills the
///     race in which an attacker who observed the URL fragment could
///     rotate the active cookie value out from under the operator.
///   - HIGH-1: constant-time token comparison via `subtle::ConstantTimeEq`
///     — Rust's `&` over `bool` is not guaranteed constant-time by the
///     optimizer; the dedicated crate is.
///   - MED-1: `Max-Age=86400` ensures browsers GC stale cookies after a
///     day even on "restore tabs"-style long-lived sessions.
///   - MED-2: require `Content-Type: application/json` (415 otherwise) so
///     a future-loosened CORS posture can't enable a CSRF `<form>` post.
async fn auth_exchange(State(state): State<AppState>, headers: HeaderMap, body: Bytes) -> Response {
    use std::sync::atomic::Ordering;
    use subtle::ConstantTimeEq;

    // MED-2 — strict Content-Type gate. The bundle's fetch always sends
    // application/json; anything else is either a misconfigured client or
    // a CSRF attempt.
    let ct_ok = headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .map(|s| {
            // accept `application/json` or `application/json; charset=...`
            let trimmed = s.split(';').next().unwrap_or("").trim();
            trimmed.eq_ignore_ascii_case("application/json")
        })
        .unwrap_or(false);
    if !ct_ok {
        let mut resp = (
            StatusCode::UNSUPPORTED_MEDIA_TYPE,
            "Content-Type must be application/json\n",
        )
            .into_response();
        resp.headers_mut().insert(
            header::CONTENT_TYPE,
            HeaderValue::from_static("text/plain; charset=utf-8"),
        );
        return resp;
    }

    // Parse the body explicitly so the Content-Type gate above runs even
    // on malformed JSON (otherwise `Json<T>` parses before the gate).
    let parsed: ExchangeRequest = match serde_json::from_slice(&body) {
        Ok(p) => p,
        Err(_) => return (StatusCode::BAD_REQUEST, "invalid json body\n").into_response(),
    };

    // CRIT-1 — single-use gate, checked BEFORE the constant-time compare
    // so we don't even leak token-shape via timing on replays.
    if state.exchange_consumed.load(Ordering::SeqCst) {
        return (
            StatusCode::UNAUTHORIZED,
            "session token already exchanged\n",
        )
            .into_response();
    }

    // HIGH-1 — constant-time compare. Token is fixed-length (44 chars of
    // URL_SAFE_NO_PAD over 32 bytes); even so, `subtle::ConstantTimeEq`
    // protects against optimizer rewrites and any future change to a
    // variable-length token format.
    let provided = parsed.sess.as_bytes();
    let expected = state.session_token.as_bytes();
    if provided.len() != expected.len() || provided.ct_eq(expected).unwrap_u8() == 0 {
        return (StatusCode::UNAUTHORIZED, "bad session token\n").into_response();
    }

    // Flip the consumed flag BEFORE minting the cookie. If two concurrent
    // requests both passed the check above, only one will see
    // `compare_exchange` return Ok — the other gets 401, preserving
    // single-use semantics under load.
    if state
        .exchange_consumed
        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
        .is_err()
    {
        return (
            StatusCode::UNAUTHORIZED,
            "session token already exchanged\n",
        )
            .into_response();
    }

    // Mint the cookie value. Distinct from the URL token so we never reuse
    // a value that might have appeared anywhere in history.
    let mut buf = [0u8; 32];
    rand::thread_rng().fill_bytes(&mut buf);
    let cookie_value = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(buf);

    {
        let mut slot = state.session_cookie.write().await;
        *slot = Some(cookie_value.clone());
    }

    // MED-1: include Max-Age so browsers GC the cookie even when the
    // process is long gone (tab restore scenario).
    let cookie_header = format!(
        "cellctl_session={}; HttpOnly; SameSite=Strict; Path=/; Max-Age={}",
        cookie_value, COOKIE_MAX_AGE_SECS
    );

    let mut resp = (StatusCode::OK, "ok\n").into_response();
    resp.headers_mut().insert(
        header::SET_COOKIE,
        HeaderValue::from_str(&cookie_header).unwrap(),
    );
    resp
}

/// Verify the request carries the `cellctl_session` cookie we set in
/// `/auth/exchange`. Returns Some(()) if OK, None if missing/mismatch.
async fn require_session_cookie(state: &AppState, headers: &HeaderMap) -> bool {
    let expected = match state.session_cookie.read().await.clone() {
        Some(v) => v,
        None => return false,
    };
    let Some(cookie_hdr) = headers.get(header::COOKIE) else {
        return false;
    };
    let Ok(cookie_str) = cookie_hdr.to_str() else {
        return false;
    };
    // Cookies are `name=value; name=value` — find our entry.
    for entry in cookie_str.split(';') {
        let entry = entry.trim();
        if let Some(v) = entry.strip_prefix("cellctl_session=") {
            return v == expected;
        }
    }
    false
}

/// `GET /v1/*` — reverse-proxy to upstream with Bearer injection.
async fn proxy_v1(
    State(state): State<AppState>,
    AxumPath(rest): AxumPath<String>,
    uri: Uri,
    headers: HeaderMap,
) -> Response {
    if !require_session_cookie(&state, &headers).await {
        return unauthorized();
    }

    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
    let upstream_url = format!("{}/v1/{}{}", state.upstream_base, rest, query);

    let client = match reqwest::Client::builder().build() {
        Ok(c) => c,
        Err(e) => return upstream_error(format!("client: {e}")),
    };

    let mut req = client.get(&upstream_url);
    if let Some(tok) = state.upstream_bearer.as_ref() {
        req = req.header(reqwest::header::AUTHORIZATION, format!("Bearer {tok}"));
    }

    let upstream_resp = match req.send().await {
        Ok(r) => r,
        Err(e) => return upstream_error(format!("send: {e}")),
    };

    let status = upstream_resp.status();
    let resp_headers = upstream_resp.headers().clone();
    let body_bytes = match upstream_resp.bytes().await {
        Ok(b) => b,
        Err(e) => return upstream_error(format!("read body: {e}")),
    };

    let mut out = Response::new(Body::from(body_bytes));
    *out.status_mut() = StatusCode::from_u16(status.as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
    // Forward content-type only (avoid leaking upstream Set-Cookie etc.).
    if let Some(ct) = resp_headers.get(reqwest::header::CONTENT_TYPE) {
        if let Ok(v) = HeaderValue::from_bytes(ct.as_bytes()) {
            out.headers_mut().insert(header::CONTENT_TYPE, v);
        }
    }
    out
}

/// `GET /ws/events` — WebSocket upgrade, forwarded to upstream.
async fn ws_events(
    State(state): State<AppState>,
    uri: Uri,
    headers: HeaderMap,
    ws: Option<WebSocketUpgrade>,
) -> Response {
    if !require_session_cookie(&state, &headers).await {
        return unauthorized();
    }
    let Some(ws) = ws else {
        return (StatusCode::BAD_REQUEST, "expected websocket upgrade\n").into_response();
    };

    let query = uri.query().map(|q| format!("?{q}")).unwrap_or_default();
    let ws_url = {
        // http:// → ws://, https:// → wss://
        let base = state.upstream_base.as_str();
        let ws_base = if let Some(rest) = base.strip_prefix("https://") {
            format!("wss://{rest}")
        } else if let Some(rest) = base.strip_prefix("http://") {
            format!("ws://{rest}")
        } else {
            base.to_string()
        };
        format!("{ws_base}/ws/events{query}")
    };
    let bearer = state.upstream_bearer.as_ref().clone();
    // Wave 2 red-team (MED-W2D-4): forward the client's
    // `Sec-WebSocket-Protocol` to the upstream. axum's
    // `WebSocketUpgrade::on_upgrade` finishes the handshake with the
    // *client* without telling us which subprotocol(s) were requested, so
    // we extract the header from the original request and propagate it
    // verbatim. The upstream's choice (if any) is returned in its 101
    // response — `connect_async` drops the response handle on the floor
    // here, which matches axum's behaviour of accepting whichever
    // subprotocol the underlying handler agreed to. If/when we want
    // strict subprotocol negotiation, the upstream's `Sec-WebSocket-Protocol`
    // header in `_response` would need to be plumbed back into the client
    // accept frame; today's contract is "best-effort forwarding".
    let subprotocols: Option<String> = headers
        .get(axum::http::header::SEC_WEBSOCKET_PROTOCOL)
        .and_then(|v| v.to_str().ok())
        .map(|s| s.to_string());

    ws.on_upgrade(move |client_ws| async move {
        let (mut client_tx, mut client_rx) = client_ws.split();

        // Build a request to upstream with the bearer header, since
        // `connect_async` accepts a raw URL but loses our custom headers.
        let mut request =
            match tokio_tungstenite::tungstenite::client::IntoClientRequest::into_client_request(
                ws_url.as_str(),
            ) {
                Ok(r) => r,
                Err(_) => return,
            };
        if let Some(tok) = bearer {
            if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(&format!(
                "Bearer {tok}"
            )) {
                request.headers_mut().insert(
                    tokio_tungstenite::tungstenite::http::header::AUTHORIZATION,
                    v,
                );
            }
        }
        if let Some(proto) = subprotocols.as_ref() {
            if let Ok(v) = tokio_tungstenite::tungstenite::http::HeaderValue::from_str(proto) {
                request.headers_mut().insert(
                    tokio_tungstenite::tungstenite::http::header::SEC_WEBSOCKET_PROTOCOL,
                    v,
                );
            }
        }
        let (upstream_ws, _) = match tokio_tungstenite::connect_async(request).await {
            Ok(p) => p,
            Err(_) => return,
        };
        let (mut up_tx, mut up_rx) = upstream_ws.split();

        loop {
            tokio::select! {
                msg = client_rx.next() => match msg {
                    Some(Ok(axum::extract::ws::Message::Text(t))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Text(t))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Binary(b))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Binary(b))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Ping(p))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Ping(p))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Pong(p))) => {
                        let _ = up_tx
                            .send(tokio_tungstenite::tungstenite::Message::Pong(p))
                            .await;
                    }
                    Some(Ok(axum::extract::ws::Message::Close(_))) | None => break,
                    Some(Err(_)) => break,
                },
                msg = up_rx.next() => match msg {
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Text(t))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Binary(b))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Binary(b))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Ping(p))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Ping(p))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Pong(p))) => {
                        let _ = client_tx
                            .send(axum::extract::ws::Message::Pong(p))
                            .await;
                    }
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) | None => break,
                    Some(Ok(tokio_tungstenite::tungstenite::Message::Frame(_))) => {}
                    Some(Err(_)) => break,
                },
            }
        }
    })
}

fn unauthorized() -> Response {
    (StatusCode::UNAUTHORIZED, "missing session cookie\n").into_response()
}

fn upstream_error(msg: String) -> Response {
    (StatusCode::BAD_GATEWAY, format!("upstream: {msg}\n")).into_response()
}

// LOW-1 / LOW-2 (red-team pass B): the dead `state_upstream_for_log` and
// `_BYTES_KEEP` shim are gone. `Bytes` is now used directly by
// `auth_exchange` (CRIT-1 / MED-2 rewrite) and the startup banner reads
// `upstream_log` directly in `run()`.

#[allow(dead_code)]
const _HEADER_NAME_KEEP: fn() = || {
    // `HeaderName` is imported for future use in custom header-injection
    // paths; keep this shim until that lands.
    let _: HeaderName = HeaderName::from_static("x-keep");
};

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::to_bytes;
    use axum::http::Request;
    use tower::ServiceExt; // for `oneshot`

    fn test_state(bundle_dir: PathBuf) -> AppState {
        AppState {
            upstream_base: Arc::new("http://127.0.0.1:0".to_string()),
            upstream_bearer: Arc::new(None),
            session_token: Arc::new("test-token".to_string()),
            session_cookie: Arc::new(RwLock::new(None)),
            exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            bundle_dir: Arc::new(bundle_dir),
        }
    }

    /// Use the real bundle if it's been built; otherwise create a fake one
    /// in a tempdir with a minimal index.html.
    fn ensure_bundle_dir() -> PathBuf {
        let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
        let real = manifest_dir.join("static");
        if real.join("index.html").exists() {
            return real;
        }
        // Fall back to a freshly-minted temp dir.
        let tmp = std::env::temp_dir().join(format!("cellctl-webui-test-{}", std::process::id()));
        std::fs::create_dir_all(&tmp).expect("mkdir tmp bundle");
        std::fs::write(
            tmp.join("index.html"),
            "<!doctype html><title>cellctl webui</title>",
        )
        .expect("write index.html");
        tmp
    }

    #[tokio::test]
    async fn serves_index_at_root() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap();
        let body_str = std::str::from_utf8(&body).unwrap();
        assert!(
            body_str.to_ascii_lowercase().contains("<!doctype html")
                || body_str.to_ascii_lowercase().contains("<html"),
            "expected HTML at /, got: {body_str:.200}"
        );
    }

    #[tokio::test]
    async fn non_get_returns_405() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("DELETE")
                    .uri("/v1/formations")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
        assert_eq!(
            resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
            Some(b"GET" as &[u8]),
        );
    }

    #[tokio::test]
    async fn put_to_v1_returns_405() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("PUT")
                    .uri("/v1/formations/foo")
                    .body(Body::from("body"))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
    }

    #[tokio::test]
    async fn proxy_without_cookie_returns_401() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_exchange_with_wrong_token_returns_401() {
        let bundle = ensure_bundle_dir();
        let app = build_router(test_state(bundle));

        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"wrong"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn auth_exchange_with_right_token_sets_cookie() {
        let bundle = ensure_bundle_dir();
        let state = test_state(bundle);
        let app = build_router(state.clone());

        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let cookie = resp
            .headers()
            .get(header::SET_COOKIE)
            .expect("Set-Cookie header present")
            .to_str()
            .unwrap()
            .to_string();
        assert!(cookie.starts_with("cellctl_session="));
        assert!(cookie.contains("HttpOnly"));
        assert!(cookie.contains("SameSite=Strict"));

        // And the state should now hold a matching cookie value.
        let stored = state.session_cookie.read().await.clone();
        assert!(stored.is_some());
        let stored = stored.unwrap();
        assert!(cookie.contains(&format!("cellctl_session={stored}")));
    }

    #[tokio::test]
    async fn proxy_with_valid_cookie_attempts_upstream() {
        // After /auth/exchange succeeds, a subsequent GET /v1/* should *not*
        // 401 — it'll instead 502 (bad gateway) because there's no real
        // upstream listening, which is exactly what proves the cookie gate
        // passed.
        let bundle = ensure_bundle_dir();
        let state = test_state(bundle);
        let app = build_router(state.clone());

        let exch = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(exch.status(), StatusCode::OK);
        let cookie_hdr = exch
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap()
            .to_string();
        // Strip attributes; keep just `cellctl_session=<value>`.
        let cookie_pair = cookie_hdr.split(';').next().unwrap().trim().to_string();

        let resp = app
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .header(header::COOKIE, cookie_pair)
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        // No live upstream → bad gateway, not 401.
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    /// Proxy invariant: no matter what HTTP method the browser sends, the
    /// upstream cellos-server must NEVER see a non-GET. The proxy is the
    /// structural enforcement of ADR-0016's read-only browser boundary
    /// (ADR-0017 §Decision 4, item 5: "Refuse non-GET methods with HTTP 405").
    ///
    /// We spin up a tiny inline mock upstream on a real loopback port that
    /// counts the methods it sees, point AppState at it, and fire a battery of
    /// inbound methods through the proxy. The assertion is bidirectional:
    ///
    ///   1. The mock saw ZERO non-GET requests.
    ///   2. Every non-GET inbound request returned 405 with `Allow: GET`.
    #[tokio::test]
    async fn proxy_only_emits_get_to_upstream() {
        use std::sync::Mutex;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpListener;

        // ---- inline mock upstream ----
        let mock = TcpListener::bind("127.0.0.1:0").await.expect("bind mock");
        let mock_addr = mock.local_addr().expect("mock addr");
        let methods_seen: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
        let methods_for_task = methods_seen.clone();

        // Spawn a tiny accept loop that reads the request line, records the
        // method, then writes a minimal `HTTP/1.1 200 OK` and closes. We do
        // not implement keep-alive, content-length parsing, or anything else
        // — the proxy doesn't keep connections open across requests against
        // this mock anyway (each `reqwest::Client::builder().build()` builds a
        // fresh client).
        tokio::spawn(async move {
            loop {
                let (mut stream, _) = match mock.accept().await {
                    Ok(p) => p,
                    Err(_) => return,
                };
                let methods = methods_for_task.clone();
                tokio::spawn(async move {
                    let mut buf = [0u8; 4096];
                    let n = match stream.read(&mut buf).await {
                        Ok(n) => n,
                        Err(_) => return,
                    };
                    let head = String::from_utf8_lossy(&buf[..n]).to_string();
                    if let Some(first_line) = head.lines().next() {
                        if let Some(method) = first_line.split_whitespace().next() {
                            methods.lock().unwrap().push(method.to_string());
                        }
                    }
                    let _ = stream
                        .write_all(
                            b"HTTP/1.1 200 OK\r\n\
                              Content-Type: application/json\r\n\
                              Content-Length: 2\r\n\
                              Connection: close\r\n\
                              \r\n\
                              {}",
                        )
                        .await;
                    let _ = stream.shutdown().await;
                });
            }
        });

        // ---- proxy under test ----
        let bundle = ensure_bundle_dir();
        let state = AppState {
            upstream_base: Arc::new(format!("http://{}", mock_addr)),
            upstream_bearer: Arc::new(Some("test-bearer".to_string())),
            session_token: Arc::new("test-token".to_string()),
            session_cookie: Arc::new(RwLock::new(None)),
            exchange_consumed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            bundle_dir: Arc::new(bundle),
        };
        let app = build_router(state.clone());

        // Mint a session cookie so /v1/* GETs can pass the auth gate. (Non-GET
        // requests will be rejected at the 405 middleware *before* the cookie
        // check, so this only matters for the GET-passes-through assertion.)
        let exch = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/auth/exchange")
                    .header(header::CONTENT_TYPE, "application/json")
                    .body(Body::from(r#"{"sess":"test-token"}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(exch.status(), StatusCode::OK);
        let cookie_pair = exch
            .headers()
            .get(header::SET_COOKIE)
            .unwrap()
            .to_str()
            .unwrap()
            .split(';')
            .next()
            .unwrap()
            .trim()
            .to_string();

        // Fire a battery of inbound methods. Each non-GET MUST return 405 and
        // MUST NOT touch the upstream.
        let non_get_methods = ["POST", "PUT", "DELETE", "PATCH"];
        for m in non_get_methods {
            let resp = app
                .clone()
                .oneshot(
                    Request::builder()
                        .method(m)
                        .uri("/v1/formations")
                        .header(header::COOKIE, cookie_pair.clone())
                        .body(Body::from("payload that must never reach upstream"))
                        .unwrap(),
                )
                .await
                .unwrap();
            assert_eq!(
                resp.status(),
                StatusCode::METHOD_NOT_ALLOWED,
                "inbound {m} should be 405"
            );
            assert_eq!(
                resp.headers().get(header::ALLOW).map(|v| v.as_bytes()),
                Some(b"GET" as &[u8]),
                "405 response for {m} must carry `Allow: GET`"
            );
        }

        // Now fire an actual GET, which SHOULD reach the mock upstream.
        let get_resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("GET")
                    .uri("/v1/formations")
                    .header(header::COOKIE, cookie_pair.clone())
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        // The GET must NOT be rejected by the proxy's read-only gate.
        assert_ne!(
            get_resp.status(),
            StatusCode::METHOD_NOT_ALLOWED,
            "GET must pass the 405 middleware"
        );

        // Give the mock a moment to record the inbound connection.
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // ---- assertions on the mock's observation log ----
        let observed = methods_seen.lock().unwrap().clone();
        // Exactly one upstream request (the GET) — the four non-GETs were
        // all stopped at the proxy.
        let non_get_count = observed.iter().filter(|m| m.as_str() != "GET").count();
        assert_eq!(
            non_get_count, 0,
            "upstream saw non-GET method(s): {observed:?}"
        );
        assert!(
            observed.iter().any(|m| m == "GET"),
            "expected at least one GET to reach upstream, saw {observed:?}"
        );
    }
}