synaps 0.11.0

//! `synaps auth-broker` — credential broker (epic #155, task #156).
//!
//! Holds ONE OAuth credential (this machine's `auth.json`), is the SINGLE
//! refresher (Anthropic rotates the refresh token on every refresh, so exactly
//! one party may refresh), and serves short-lived **access** tokens to authorized
//! client machines.
//!
//! ## Transport security (task #156 subtask 6)
//!
//! ```text
//! bind addr    | --tls-cert/key | --insecure-http | result
//! -------------|----------------|-----------------|-------------------------------------------
//! loopback     | absent         | any             | plain HTTP, no warning
//! loopback     | both set       | any             | HTTPS (redundant but allowed)
//! non-loopback | absent         | absent          | REFUSED — must TLS or explicit opt-in
//! non-loopback | absent         | present         | plain HTTP + WireGuard guidance warning
//! non-loopback | both set       | any             | HTTPS, clean startup
//! one of pair  | one set        | any             | REFUSED — must provide both cert+key
//! ```
//!
//! TLS is served via `axum-server` (tls-rustls feature, rustls 0.23).
//!
//! ## Machine auth (#147 C1)
//!
//! Every credential endpoint requires `Authorization: Bearer <machine token>`.
//! The token comes from `--machine-token-file` or `SYNAPS_BROKER_TOKEN`. A
//! loopback bind with neither generates one into `<base_dir>/auth-broker.token`
//! (0600, next to `auth.json`; reused on restart), and a local client whose
//! endpoint is a loopback URL reads that file when it has no
//! `auth.machine_token` / `SYNAPS_MACHINE_TOKEN`. A loopback broker also
//! refuses any `Host` other than `127.0.0.1:<port>`, `localhost:<port>` or
//! `[::1]:<port>` (DNS rebinding). Only a non-loopback bind with
//! `--insecure-no-auth` runs without a token.
//!
//! Clients run `synaps` with `auth.remote_endpoint` pointed here; they fetch a
//! token per request and never store the credential on their own disk.
//!
//! Endpoints:
//!   GET  /healthz                         -> { status }   (no secret; 503 until ANY OAuth slot exists)
//!   GET  /token?provider=X[&account=Y]    -> { access_token, expires, ttl_ms, account }
//!                                            (machine-auth, OAuth providers ONLY, never Muse
//!                                            (proxy-only, #216 I2); `account` is a
//!                                            label, `default`, or `auto`; unknown → 404, malformed
//!                                            → 400, no capacity → 503; never a fallback)
//!   POST /proxy                           -> typed broker proxy (machine-auth; OAuth providers may
//!                                            pin a slot as `provider@label`; keys never vended)
//!   GET  /usage                           -> legacy raw Anthropic usage JSON (machine-auth)
//!   GET  /usage/snapshot?provider&account -> normalized UsageSnapshot (machine-auth)
//!   POST /accounts/cooldown               -> { provider, account, until_ms?, reason } (machine-auth)
//!   GET  /capabilities                    -> provider status list incl. account rows (no secrets)

use std::net::SocketAddr;
use std::path::PathBuf;
use std::time::Duration;

use std::sync::Arc;

use axum::{
    extract::{ConnectInfo, Query, State},
    http::{HeaderMap, StatusCode},
    response::IntoResponse,
    routing::{get, post},
    Json, Router,
};
use futures::StreamExt;
use serde::Deserialize;
use serde_json::json;
use synaps_cli::auth;

/// OAuth providers whose descriptors permit access-token vending. Static-key
/// strategies are intentionally not served by this endpoint.
fn broker_provider(value: &str) -> Option<auth::OAuthProviderId> {
    let id: auth::OAuthProviderId = value.try_into().ok()?;
    let descriptor = auth::provider::registry().get(id).copied()?;
    (descriptor.broker_strategy == auth::BrokerCredentialStrategy::OAuthAccessToken).then_some(id)
}

/// Providers `/token` may vend an access token for: [`broker_provider`]
/// minus Meta Muse (#216 I2). Muse is routed `BrokerProxy` and its stored
/// `access` is a minted Model API key that is valid until it 401s, not a
/// short-lived OAuth access token, so it never leaves the broker. `/proxy`
/// (which does not consult this list) keeps serving Muse, and usage /
/// cooldown endpoints keep using `broker_provider`.
fn token_vending_provider(value: &str) -> Option<auth::OAuthProviderId> {
    broker_provider(value).filter(|id| *id != auth::OAuthProviderId::Muse)
}

// ── TLS config types ─────────────────────────────────────────────────────────

/// Validated, ready-to-use TLS material (PEM bytes confirmed parseable).
#[derive(Clone, Debug)]
pub struct TlsConfig {
    pub cert_pem: Vec<u8>,
    pub key_pem: Vec<u8>,
}

/// Outcome of [`check_bind_policy`] — what transport mode to use or why to abort.
#[derive(Debug, PartialEq, Eq)]
pub enum BindDecision {
    /// Plain HTTP is allowed (loopback or explicit opt-in on non-loopback).
    Http {
        /// Print the WireGuard guidance warning (non-loopback + --insecure-http).
        warn: bool,
    },
    /// Serve TLS.
    Tls,
    /// Refuse to start; contains the error message.
    Refuse(String),
}

// ── Bind-policy logic (pure — trivial to unit test) ──────────────────────────

/// Pure function — no I/O. Determines whether to serve HTTP, HTTPS, or refuse.
///
/// `is_loopback` — true if the bind address is 127.x or ::1  
/// `has_tls`     — both --tls-cert and --tls-key were provided  
/// `insecure_http` — --insecure-http flag was passed
pub fn check_bind_policy(is_loopback: bool, has_tls: bool, insecure_http: bool) -> BindDecision {
    if has_tls {
        return BindDecision::Tls;
    }
    if is_loopback {
        return BindDecision::Http { warn: false };
    }
    // non-loopback, no TLS
    if insecure_http {
        return BindDecision::Http { warn: true };
    }
    BindDecision::Refuse(
        "refusing to start: non-loopback bind without TLS. \
         Pass --tls-cert + --tls-key to enable HTTPS (recommended), \
         or --insecure-http to acknowledge you are running behind WireGuard \
         / a private network and accept plaintext."
            .to_string(),
    )
}

// ── PEM validation ────────────────────────────────────────────────────────────

/// Read and validate a cert+key PEM pair. Fails fast with a human-readable error
/// so the broker never starts with a broken TLS config.
pub fn load_and_validate_tls(cert_path: &PathBuf, key_path: &PathBuf) -> anyhow::Result<TlsConfig> {
    // Read files first — surface "file not found" before doing any parsing.
    let cert_pem = std::fs::read(cert_path)
        .map_err(|e| anyhow::anyhow!("--tls-cert '{}': {e}", cert_path.display()))?;
    let key_pem = std::fs::read(key_path)
        .map_err(|e| anyhow::anyhow!("--tls-key '{}': {e}", key_path.display()))?;

    // Parse to catch garbage PEM / wrong file type before binding.
    validate_pem_pair(&cert_pem, &key_pem)?;

    Ok(TlsConfig { cert_pem, key_pem })
}

/// Parse the PEM bytes — returns a readable error if either is malformed.
/// Uses `rustls-pki-types` (same parser axum-server uses internally).
/// Separated for unit-testability with in-memory bytes.
pub fn validate_pem_pair(cert_pem: &[u8], key_pem: &[u8]) -> anyhow::Result<()> {
    use rustls_pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer};

    // Validate cert chain — must have at least one cert.
    let certs: Vec<CertificateDer<'_>> = CertificateDer::pem_slice_iter(cert_pem)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| anyhow::anyhow!("TLS cert PEM is malformed: {e}"))?;
    if certs.is_empty() {
        anyhow::bail!("TLS cert PEM contains no certificates (empty or wrong format)");
    }

    // Validate private key — must parse as one of the known key types.
    let keys: Vec<PrivateKeyDer<'_>> = PrivateKeyDer::pem_slice_iter(key_pem)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| anyhow::anyhow!("TLS key PEM is malformed: {e}"))?;
    if keys.is_empty() {
        anyhow::bail!("TLS key PEM contains no private key (empty or wrong format)");
    }
    if keys.len() > 1 {
        anyhow::bail!("TLS key PEM contains multiple private keys; it must contain exactly one");
    }

    Ok(())
}

// ── Broker internal state ─────────────────────────────────────────────────────

#[derive(Clone)]
struct BrokerState {
    /// Token clients must present as `Authorization: Bearer <token>`. `None`
    /// disables auth: only a non-loopback bind with explicit
    /// `--insecure-no-auth`. A loopback broker always has one (#147 C1).
    machine_token: Option<String>,
    /// Loopback binds only: the hostnames a `Host` header may name
    /// (`localhost`, `127.0.0.1`, `[::1]` and the bound IP), any port.
    /// Anything else is refused before routing, so a DNS-rebinding page
    /// (`Host: attacker.example`) cannot talk to the broker through the
    /// victim's browser. The port is deliberately not checked: it adds no
    /// protection and would break `ssh -L <other-port>:127.0.0.1:<port>`.
    allowed_hosts: Option<Arc<[String]>>,
    /// Where the expected machine token comes from, for actionable denial
    /// logs (a path or an env var name; never the token).
    token_source: Arc<str>,
    /// The in-process broker that owns every credential (all providers and
    /// account slots), refreshes them, and executes proxied requests. Raw
    /// keys and refresh tokens never leave this boundary.
    local: Arc<auth::LocalBroker>,
}

/// Hostnames a loopback broker bound to `bind_ip` accepts in `Host`
/// (compared without the port).
fn loopback_allowed_hostnames(bind_ip: std::net::IpAddr) -> Vec<String> {
    let mut names: Vec<String> = ["localhost", "127.0.0.1", "[::1]"]
        .iter()
        .map(|n| n.to_string())
        .collect();
    let bound = match bind_ip {
        std::net::IpAddr::V4(ip) => ip.to_string(),
        std::net::IpAddr::V6(ip) => format!("[{ip}]"),
    };
    if !names.contains(&bound) {
        names.push(bound);
    }
    names
}

/// The hostname part of a `Host` / `:authority` value, if well-formed.
fn host_header_hostname(value: &str) -> Option<String> {
    let authority: axum::http::uri::Authority = value.parse().ok()?;
    if authority.as_str().contains('@') {
        return None; // userinfo has no place in a Host header
    }
    Some(authority.host().to_string())
}

/// Refuse any request whose `Host` (or HTTP/2 `:authority`) is not on the
/// loopback allowlist. No-op for non-loopback binds.
async fn host_guard(
    State(st): State<BrokerState>,
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    if let Some(allowed) = st.allowed_hosts.as_deref() {
        let host = req
            .headers()
            .get(axum::http::header::HOST)
            .and_then(|v| v.to_str().ok())
            .or_else(|| req.uri().authority().map(|a| a.as_str()))
            .unwrap_or("");
        let hostname = host_header_hostname(host);
        if !hostname
            .as_deref()
            .is_some_and(|h| allowed.iter().any(|a| a.eq_ignore_ascii_case(h)))
        {
            eprintln!(
                "[auth-broker] DENIED request with non-loopback Host ({} chars)",
                host.len()
            );
            return (
                StatusCode::FORBIDDEN,
                Json(json!({ "error": "host not allowed" })),
            )
                .into_response();
        }
    }
    next.run(req).await
}

/// The machine token this broker will require. An explicit token (file or
/// env) wins. A loopback bind with none uses (creating on first start) the
/// 0600 token file at `token_path`, which local clients read automatically;
/// `--insecure-no-auth` does not turn auth off on loopback. A non-loopback
/// bind with none is refused unless `insecure_no_auth`.
fn resolve_machine_token(
    explicit: Option<String>,
    is_loopback: bool,
    insecure_no_auth: bool,
    token_path: &std::path::Path,
) -> anyhow::Result<Option<String>> {
    if explicit.is_some() {
        return Ok(explicit);
    }
    if is_loopback {
        if insecure_no_auth {
            eprintln!(
                "[auth-broker] WARNING: --insecure-no-auth is ignored on a loopback bind; \
                 a loopback broker always requires a machine token"
            );
        }
        let (token, origin) = auth::load_or_create_broker_token(token_path).map_err(|e| {
            anyhow::anyhow!("broker token file {}: {e}", token_path.display())
        })?;
        match origin {
            auth::BrokerTokenOrigin::Rotated => eprintln!(
                "[auth-broker] WARNING: {} was readable by other users, so its token may have \
                 leaked. Generated a NEW token (0600). Restart or reload every client that \
                 used the old one.",
                token_path.display()
            ),
            auth::BrokerTokenOrigin::Created | auth::BrokerTokenOrigin::Loaded => eprintln!(
                "[auth-broker] loopback machine token {} {} (0600; local clients read it automatically)",
                if origin == auth::BrokerTokenOrigin::Created {
                    "generated at"
                } else {
                    "loaded from"
                },
                token_path.display()
            ),
        }
        return Ok(Some(token));
    }
    if !insecure_no_auth {
        anyhow::bail!(
            "refusing to start: no machine token (set --machine-token-file / \
             SYNAPS_BROKER_TOKEN) while bound to a non-loopback address. That would serve \
             credentials unauthenticated to the network. Pass --insecure-no-auth to override, \
             or bind 127.0.0.1."
        );
    }
    Ok(None)
}

/// 401 for a request that failed machine auth, with an actionable log line
/// (prA I2): old clients can't explain a bare 401, so the broker does. The
/// token itself is never logged or echoed.
fn deny_machine_auth(
    st: &BrokerState,
    headers: &HeaderMap,
    route: &str,
    peer: Option<std::net::IpAddr>,
) -> axum::response::Response {
    let presented = headers.get("authorization").is_some();
    let peer_label = peer.map_or_else(|| "a client".to_string(), |ip| ip.to_string());
    // Rate-limited (prA R2-m6) per (reason, route, peer): a retrying old
    // client can't flood the log, and each route still names its peer.
    let key = denial_key(presented, route, peer);
    if let Some(suppressed) = denial_advisory_due(&DENIAL_LOG, &key, std::time::Instant::now(), DENIAL_LOG_EVERY) {
        let also = if suppressed > 0 {
            format!(" ({suppressed} similar denials suppressed in the last minute)")
        } else {
            String::new()
        };
        if presented {
            eprintln!(
                "[auth-broker] DENIED {route} from {peer_label} (wrong machine token){also}. The \
                 expected token is {}; if it was regenerated, restart or reload the client.",
                st.token_source
            );
        } else {
            eprintln!(
                "[auth-broker] DENIED {route} from {peer_label} (no machine token presented){also}. \
                 Clients must set auth.machine_token or SYNAPS_MACHINE_TOKEN to the token in {}. \
                 Clients older than this broker, or running as another user or base dir, do not \
                 pick it up automatically; restart or reload clients after it changes.",
                st.token_source
            );
        }
    }
    (
        StatusCode::UNAUTHORIZED,
        Json(json!({
            "error": "bad machine auth",
            "hint": "set auth.machine_token or SYNAPS_MACHINE_TOKEN to the broker's machine token \
                     (a loopback broker writes it to auth-broker.token in its base dir); restart \
                     or reload the client after it changes",
        })),
    )
        .into_response()
}

/// Rate-limit key for a 401 advisory: (missing|wrong token, route, peer).
fn denial_key(presented: bool, route: &str, peer: Option<std::net::IpAddr>) -> String {
    let reason = if presented { "wrong" } else { "missing" };
    let peer = peer.map_or_else(|| "-".to_string(), |ip| ip.to_string());
    format!("{reason}|{route}|{peer}")
}

/// One advisory per (reason, route, peer) per this interval.
const DENIAL_LOG_EVERY: std::time::Duration = std::time::Duration::from_secs(60);

type DenialLog = std::sync::Mutex<std::collections::HashMap<String, (std::time::Instant, u64)>>;

static DENIAL_LOG: std::sync::LazyLock<DenialLog> = std::sync::LazyLock::new(Default::default);

/// Whether the advisory for `key` should be logged now. `Some(n)`: log it,
/// `n` similar denials were suppressed since the last one; `None`: suppress.
fn denial_advisory_due(
    log: &DenialLog,
    key: &str,
    now: std::time::Instant,
    every: std::time::Duration,
) -> Option<u64> {
    let mut log = log.lock().unwrap_or_else(|e| e.into_inner());
    if log.len() > 1024 {
        log.retain(|_, (at, _)| now.duration_since(*at) < every);
    }
    match log.get_mut(key) {
        Some((at, suppressed)) if now.duration_since(*at) < every => {
            *suppressed += 1;
            None
        }
        Some((at, suppressed)) => {
            let n = *suppressed;
            *at = now;
            *suppressed = 0;
            Some(n)
        }
        None => {
            log.insert(key.to_string(), (now, 0));
            Some(0)
        }
    }
}

/// Constant-time machine-auth check shared by every credential endpoint.
fn machine_auth_ok(st: &BrokerState, headers: &HeaderMap) -> bool {
    let Some(ref expected) = st.machine_token else {
        return true; // auth disabled (loopback / explicit insecure opt-in)
    };
    let got = headers
        .get("authorization")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    let expected_header = format!("Bearer {expected}");
    ct_eq(got.as_bytes(), expected_header.as_bytes())
}

#[derive(Deserialize)]
struct TokenQuery {
    provider: Option<String>,
    /// `default`, a label, or `auto`. Absent → the broker host's policy.
    account: Option<String>,
    /// Model hint for capacity-aware `auto` selection.
    model: Option<String>,
}

#[derive(Deserialize)]
struct UsageSnapshotQuery {
    provider: Option<String>,
    account: Option<String>,
}

#[derive(Deserialize)]
struct CooldownReport {
    provider: String,
    account: String,
    #[serde(default)]
    until_ms: Option<u64>,
    #[serde(default)]
    reason: Option<String>,
}

/// Constant-time byte comparison — avoids a timing oracle on the machine token.
/// (#158 B1 / CWE-208.) Length mismatch short-circuits; token length is not the
/// secret. Equal-length inputs are compared in constant time.
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

// ── Entry point ───────────────────────────────────────────────────────────────

#[allow(clippy::too_many_arguments)]
pub async fn run(
    bind: String,
    machine_token: Option<String>,
    machine_token_file: Option<PathBuf>,
    insecure_no_auth: bool,
    tls_cert: Option<PathBuf>,
    tls_key: Option<PathBuf>,
    insecure_http: bool,
) -> anyhow::Result<()> {
    // ── TLS arg validation: must supply both or neither ──
    let tls = match (tls_cert, tls_key) {
        (Some(cert), Some(key)) => {
            let cfg = load_and_validate_tls(&cert, &key)?;
            Some(cfg)
        }
        (Some(_), None) => anyhow::bail!(
            "--tls-cert was given but --tls-key is missing. Provide both to enable TLS."
        ),
        (None, Some(_)) => anyhow::bail!(
            "--tls-key was given but --tls-cert is missing. Provide both to enable TLS."
        ),
        (None, None) => None,
    };

    // B3 (security): the raw `--machine-token` flag is REJECTED outright. A
    // secret passed in argv is world-readable through `ps aux` and
    // /proc/<pid>/cmdline for the whole life of the process, so any local user
    // can lift the broker credential — which fronts OAuth tokens for every
    // configured provider. The flag is still parsed so we can emit this
    // migration error instead of clap's opaque "unexpected argument".
    // Token sources are file > env only.
    if machine_token.is_some() {
        anyhow::bail!(
            "--machine-token is no longer accepted: a token in argv is visible to every local \
             user via `ps aux` and /proc/<pid>/cmdline, which leaks the broker credential.\n\
             \n\
             Use one of these instead:\n\
             \x20 --machine-token-file <path>   (file readable only by this user)\n\
             \x20 SYNAPS_BROKER_TOKEN=<token>   (environment variable)"
        );
    }

    let machine_token = match machine_token_file {
        Some(ref path) => Some(
            std::fs::read_to_string(path)
                .map_err(|e| {
                    anyhow::anyhow!(
                        "could not read --machine-token-file {}: {e}",
                        path.display()
                    )
                })?
                .trim()
                .to_string(),
        ),
        None => std::env::var("SYNAPS_BROKER_TOKEN").ok(),
    }
    .filter(|s| !s.trim().is_empty());

    let addr: SocketAddr = bind
        .parse()
        .map_err(|e| anyhow::anyhow!("invalid --bind '{}': {}", bind, e))?;

    // ── Bind policy ──────────────────────────────────────────────────────────
    let decision = check_bind_policy(addr.ip().is_loopback(), tls.is_some(), insecure_http);
    match &decision {
        BindDecision::Refuse(msg) => anyhow::bail!("{}", msg),
        BindDecision::Http { warn: true } => {
            eprintln!(
                "  ⚠  non-loopback plain-HTTP bind on {addr}.\n\
                 \n\
                 You have acknowledged this with --insecure-http. Ensure this broker is\n\
                 ONLY reachable via WireGuard or another encrypted private network layer.\n\
                 TLS in-process is available: pass --tls-cert <path> --tls-key <path>.\n"
            );
        }
        BindDecision::Http { warn: false } | BindDecision::Tls => {}
    }

    // B2 + #147 C1: a loopback broker always requires a machine token
    // (auto-generated next to auth.json when none is configured); a
    // non-loopback bind without one is refused unless explicitly opted in.
    let local_token_path = auth::local_broker_token_path();
    let token_source: Arc<str> = match (&machine_token_file, machine_token.is_some()) {
        (Some(path), _) => format!("the --machine-token-file {}", path.display()).into(),
        (None, true) => "SYNAPS_BROKER_TOKEN on the broker host".into(),
        (None, false) => format!("{} on the broker host", local_token_path.display()).into(),
    };
    let machine_token = resolve_machine_token(
        machine_token,
        addr.ip().is_loopback(),
        insecure_no_auth,
        &local_token_path,
    )?;

    // D1: fail fast if no OAuth credential of ANY provider/account is
    // present or the store is unreadable. Codex-only, Kimi-only, or
    // named-slot-only installations are valid broker hosts.
    match auth::any_oauth_credential_present() {
        Ok(true) => {}
        Ok(false) => anyhow::bail!(
            "no OAuth credential at {}. Run `synaps login --provider <id> [--account <label>]` on the broker host first.",
            auth::auth_file_path().display()
        ),
        Err(e) => anyhow::bail!("credential unreadable/corrupt: {e}"),
    }

    let client = reqwest::Client::builder()
        .tls_built_in_webpki_certs(true)
        .tls_built_in_native_certs(true)
        .connect_timeout(Duration::from_secs(10))
        .timeout(Duration::from_secs(60))
        .build()?;

    let state = BrokerState {
        machine_token: machine_token.clone(),
        allowed_hosts: addr
            .ip()
            .is_loopback()
            .then(|| Arc::from(loopback_allowed_hostnames(addr.ip()))),
        token_source,
        local: Arc::new(auth::LocalBroker::new(client.clone())),
    };

    // Proactive refresh of EVERY stored OAuth slot, SUPERVISED. Each slot
    // refreshes independently (own gate + cross-process lock); one failing
    // account never blocks the others. Log lines carry labels only.
    {
        let refresh_client = client;
        let handle = tokio::spawn(async move {
            let mut tick = tokio::time::interval(Duration::from_secs(60));
            loop {
                tick.tick().await;
                let accounts = match auth::list_all_accounts() {
                    Ok(list) => list,
                    Err(e) => {
                        eprintln!("[auth-broker] proactive refresh: cannot list accounts: {e}");
                        continue;
                    }
                };
                for summary in accounts {
                    let Some(cred) = summary.credential_ref() else {
                        continue;
                    };
                    if let Err(e) = auth::ensure_fresh_credential(&refresh_client, &cred).await {
                        eprintln!("[auth-broker] proactive refresh failed for {cred}: {e}");
                    }
                }
            }
        });
        tokio::spawn(async move {
            if let Err(e) = handle.await {
                eprintln!("[auth-broker] FATAL: proactive refresh task died: {e:?} — tokens will go stale; restart the broker.");
            }
        });
    }

    let app = build_router(state);

    match decision {
        BindDecision::Tls => {
            eprintln!(
                "synaps auth-broker listening on https://{addr}  (machine auth: {}, TLS: ON)",
                if machine_token.is_some() {
                    "ON"
                } else {
                    "OFF — --insecure-no-auth!"
                }
            );
            let tls_cfg = tls.expect("Tls decision requires tls config");
            // rustls 0.23 requires an explicit CryptoProvider when multiple backends
            // are compiled in; install ring as the default (idempotent — Err means already set).
            let _ = rustls::crypto::ring::default_provider().install_default();
            let rustls_config =
                axum_server::tls_rustls::RustlsConfig::from_pem(tls_cfg.cert_pem, tls_cfg.key_pem)
                    .await
                    .map_err(|e| anyhow::anyhow!("failed to build TLS config from PEM: {e}"))?;

            // Handle<SocketAddr> — matches addr type for bind_rustls.
            let handle = axum_server::Handle::<SocketAddr>::new();
            let h2 = handle.clone();
            tokio::spawn(async move {
                shutdown_signal().await;
                eprintln!("[auth-broker] shutting down (draining in-flight requests)…");
                h2.graceful_shutdown(Some(Duration::from_secs(30)));
            });

            axum_server::bind_rustls(addr, rustls_config)
                .handle(handle)
                .serve(app.into_make_service_with_connect_info::<SocketAddr>())
                .await
                .map_err(|e| anyhow::anyhow!("axum-server TLS serve error: {e}"))?;
        }
        BindDecision::Http { .. } => {
            eprintln!(
                "synaps auth-broker listening on http://{addr}  (machine auth: {})",
                if machine_token.is_some() {
                    "ON"
                } else {
                    "OFF — --insecure-no-auth!"
                }
            );
            let listener = tokio::net::TcpListener::bind(addr).await?;
            axum::serve(
                listener,
                app.into_make_service_with_connect_info::<SocketAddr>(),
            )
            .with_graceful_shutdown(shutdown_signal())
            .await?;
        }
        BindDecision::Refuse(_) => unreachable!("Refuse is handled above"),
    }

    Ok(())
}

/// Build the production auth-broker router with an injected typed cloud backend.
/// This is the supported construction seam for unattended production-router tests;
/// credentials and signing remain behind `CloudBackend`.
#[cfg(test)]
fn build_router_with_cloud_backend(
    machine_token: Option<String>,
    backend: Arc<dyn auth::broker::CloudBackend>,
) -> Router {
    let local = auth::LocalBroker::new(reqwest::Client::new()).with_cloud_backend(backend);
    build_router(BrokerState {
        machine_token,
        allowed_hosts: None,
        token_source: Arc::from("test"),
        local: Arc::new(local),
    })
}

/// Build the shared axum Router. Extracted so tests can reuse it.
fn build_router(state: BrokerState) -> Router {
    Router::new()
        .route("/healthz", get(healthz))
        .route("/token", get(token))
        .route("/proxy", post(proxy))
        .route("/usage", get(usage))
        .route("/usage/snapshot", get(usage_snapshot))
        .route("/accounts/cooldown", post(accounts_cooldown))
        .route("/cloud/catalog", post(cloud_catalog))
        .route("/cloud/invoke", post(cloud_invoke))
        .route("/capabilities", get(capabilities))
        // Bound inbound request bodies to the broker's typed proxy limit.
        .layer(axum::extract::DefaultBodyLimit::max(
            auth::MAX_PROXY_REQUEST_BYTES,
        ))
        // B7: cap concurrent in-flight requests.
        .layer(tower::limit::ConcurrencyLimitLayer::new(64))
        // #147 C1: loopback Host allowlist runs before every route.
        .layer(axum::middleware::from_fn_with_state(state.clone(), host_guard))
        .with_state(state)
}

/// Resolve on Ctrl-C or SIGTERM so the broker drains cleanly.
async fn shutdown_signal() {
    let ctrl_c = async {
        let _ = tokio::signal::ctrl_c().await;
    };
    #[cfg(unix)]
    let terminate = async {
        match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
            Ok(mut sig) => {
                sig.recv().await;
            }
            Err(_) => std::future::pending::<()>().await,
        }
    };
    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();
    tokio::select! {
        _ = ctrl_c => {}
        _ = terminate => {}
    }
}

// ── Handlers ──────────────────────────────────────────────────────────────────

/// Liveness + credential readiness. Returns non-200 when the credential is
/// missing/corrupt so monitors actually catch it.
async fn healthz() -> impl IntoResponse {
    match auth::any_oauth_credential_present() {
        Ok(true) => (StatusCode::OK, Json(json!({ "status": "ok" }))),
        Ok(false) => (
            StatusCode::SERVICE_UNAVAILABLE,
            Json(json!({ "status": "no_credential" })),
        ),
        Err(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({ "status": "error" })),
        ),
    }
}

/// Issue a current access token for `provider` (default: anthropic, allowlisted).
async fn token(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Query(q): Query<TokenQuery>,
) -> impl IntoResponse {
    // ── machine auth (constant-time) ──
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/token", Some(peer.ip()));
    }

    // ── provider allowlist ──
    let provider = q.provider.unwrap_or_else(|| "anthropic".to_string());
    let Some(provider_id) = token_vending_provider(&provider) else {
        eprintln!(
            "[auth-broker] DENIED from {} (provider not allowed, {} chars)",
            peer.ip(),
            provider.len()
        );
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": "unknown provider" })),
        )
            .into_response();
    };

    // ── account selection ──
    // Absent → this host's policy. `auto` → capacity selection here (the
    // client's own policy said auto). Anything else must be `default` or a
    // valid label; an unknown slot is 404 — never served from another slot.
    use synaps_cli::auth::CredentialBroker;
    let model = q.model.as_deref().filter(|m| !m.trim().is_empty());
    let pinned = match q.account.as_deref() {
        None => st.local.access_token_pinned_for(provider_id, model).await,
        Some(auth::AUTO_ACCOUNT_NAME) => st.local.access_token_auto(provider_id, model).await,
        Some(raw) => match auth::Account::parse(raw) {
            Ok(account) => {
                let cred = auth::CredentialRef::new(provider_id, account);
                st.local
                    .access_token_for(&cred)
                    .await
                    .map(|token| auth::PinnedToken {
                        credential: cred,
                        token,
                    })
            }
            Err(e) => Err(auth::BrokerError::InvalidAccount(e)),
        },
    };

    match pinned {
        Ok(p) => {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis() as u64)
                .unwrap_or(0);
            let ttl_ms = p.token.expires.saturating_sub(now);
            eprintln!(
                "[auth-broker] issued {} token to {} (expires {})",
                p.credential,
                peer.ip(),
                p.token.expires
            );
            (
                StatusCode::OK,
                Json(json!({
                    "access_token": p.token.token,
                    "expires": p.token.expires,
                    "ttl_ms": ttl_ms,
                    "account": p.credential.account.label_str(),
                })),
            )
                .into_response()
        }
        Err(e) => {
            // BrokerError Display is secret-free by contract; the wire body
            // stays generic for credential failures.
            eprintln!(
                "[auth-broker] token request for {provider} from {} failed: {}",
                peer.ip(),
                e
            );
            match e {
                auth::BrokerError::InvalidAccount(_) | auth::BrokerError::UnsupportedAccount { .. } => (
                    StatusCode::BAD_REQUEST,
                    Json(json!({ "error": "invalid account" })),
                )
                    .into_response(),
                auth::BrokerError::UnknownAccount { .. } => (
                    StatusCode::NOT_FOUND,
                    Json(json!({ "error": "unknown account" })),
                )
                    .into_response(),
                auth::BrokerError::NoAccountAvailable { .. } => (
                    StatusCode::SERVICE_UNAVAILABLE,
                    Json(json!({ "error": "no account available" })),
                )
                    .into_response(),
                _ => (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    Json(json!({ "error": "token refresh failed" })),
                )
                    .into_response(),
            }
        }
    }
}

/// Typed broker proxy: execute a static-key provider request broker-side.
///
/// OAuth providers are structurally excluded (`ProxyRequest::validate`), so
/// this endpoint can never become a second token-vending path, and no raw
/// static key ever appears in a response — the broker attaches it upstream.
async fn proxy(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Json(request): Json<auth::ProxyRequest>,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/proxy", Some(peer.ip()));
    }
    if let Err(e) = request.validate() {
        // BrokerError Display is secret-free by contract.
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": e.to_string() })),
        )
            .into_response();
    }
    use synaps_cli::auth::CredentialBroker;
    if request.stream {
        match st.local.proxy_stream(request).await {
            Ok(stream) => {
                let body = axum::body::Body::from_stream(
                    stream.map(|chunk| chunk.map_err(|e| std::io::Error::other(e.to_string()))),
                );
                (
                    StatusCode::OK,
                    [("content-type", "text/event-stream")],
                    body,
                )
                    .into_response()
            }
            Err(e) => broker_error_response(e),
        }
    } else {
        match st.local.proxy(request).await {
            Ok(resp) => (StatusCode::OK, Json(resp)).into_response(),
            Err(e) => broker_error_response(e),
        }
    }
}

/// Typed usage operation: the local broker resolves the Anthropic OAuth token
/// behind the boundary and calls the pinned usage URL; clients receive usage
/// JSON only. No token, refresh token, or URL choice ever crosses this
/// endpoint — it is deliberately not a general OAuth proxy.
async fn usage(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/usage", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    match st.local.anthropic_usage().await {
        Ok(value) => (StatusCode::OK, Json(value)).into_response(),
        Err(e) => broker_error_response(e),
    }
}

/// Normalized, secret-free usage snapshot for one `(provider, account)`.
/// The token is resolved broker-side. `account` absent → this host's
/// explicit policy selection (an `auto` policy is not a snapshot target).
async fn usage_snapshot(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Query(q): Query<UsageSnapshotQuery>,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/usage/snapshot", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    let provider = q.provider.unwrap_or_else(|| "anthropic".to_string());
    let Some(provider_id) = broker_provider(&provider) else {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": "unknown provider" })),
        )
            .into_response();
    };
    let account = match q.account.as_deref() {
        Some(raw) => match auth::Account::parse(raw) {
            Ok(account) => account,
            Err(_) => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(json!({ "error": "invalid account" })),
                )
                    .into_response()
            }
        },
        None => match st.local.account_selector(provider_id) {
            auth::AccountSelector::Account(account) => account,
            _ => {
                return (
                    StatusCode::BAD_REQUEST,
                    Json(json!({ "error": "account required" })),
                )
                    .into_response()
            }
        },
    };
    let cred = auth::CredentialRef::new(provider_id, account);
    match st.local.usage(&cred).await {
        Ok(snapshot) => (StatusCode::OK, Json(snapshot)).into_response(),
        Err(e) => broker_error_response(e),
    }
}

/// A client reports a provider limit for one account so `auto` selection on
/// this host stops advertising it. Machine-auth; labels only.
async fn accounts_cooldown(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Json(report): Json<CooldownReport>,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/accounts/cooldown", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    let Some(provider_id) = broker_provider(&report.provider) else {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({ "error": "unknown provider" })),
        )
            .into_response();
    };
    let account = match auth::Account::parse(&report.account) {
        Ok(account) => account,
        Err(_) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({ "error": "invalid account" })),
            )
                .into_response()
        }
    };
    let cred = auth::CredentialRef::new(provider_id, account);
    let reason = report.reason.as_deref().unwrap_or("client_report");
    match st.local.report_cooldown(&cred, report.until_ms, reason).await {
        Ok(()) => (StatusCode::OK, Json(json!({ "status": "ok" }))).into_response(),
        Err(e) => broker_error_response(e),
    }
}

#[derive(Deserialize)]
struct CloudCatalogRequest {
    provider: auth::CloudProviderId,
    context_ref: String,
    #[serde(default)]
    allow_stale: bool,
}

async fn cloud_catalog(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Json(request): Json<CloudCatalogRequest>,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/cloud/catalog", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    match st
        .local
        .cloud_catalog(request.provider, &request.context_ref, request.allow_stale)
        .await
    {
        Ok(entries) => (StatusCode::OK, Json(entries)).into_response(),
        Err(error) => broker_error_response(error),
    }
}

#[derive(Deserialize)]
struct CloudInvokeRequest {
    provider: auth::CloudProviderId,
    context_ref: String,
    model_id: String,
    request: auth::InvokeRequest,
}

async fn cloud_invoke(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
    Json(request): Json<CloudInvokeRequest>,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/cloud/invoke", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    match st
        .local
        .cloud_invoke(
            request.provider,
            &request.context_ref,
            &request.model_id,
            request.request,
        )
        .await
    {
        Ok(stream) => {
            // NDJSON framing survives arbitrary HTTP chunk boundaries.
            let body = axum::body::Body::from_stream(stream.map(|event| {
                event
                    .and_then(|event| {
                        serde_json::to_vec(&event)
                            .map(|mut bytes| {
                                bytes.push(b'\n');
                                bytes::Bytes::from(bytes)
                            })
                            .map_err(|e| auth::BrokerError::Transport(e.to_string()))
                    })
                    .map_err(|e| std::io::Error::other(e.to_string()))
            }));
            (
                StatusCode::OK,
                [("content-type", "application/x-ndjson")],
                body,
            )
                .into_response()
        }
        Err(error) => broker_error_response(error),
    }
}

/// Non-secret provider capability/status list.
async fn capabilities(
    State(st): State<BrokerState>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    headers: HeaderMap,
) -> axum::response::Response {
    if !machine_auth_ok(&st, &headers) {
        return deny_machine_auth(&st, &headers, "/capabilities", Some(peer.ip()));
    }
    use synaps_cli::auth::CredentialBroker;
    match st.local.capabilities().await {
        Ok(caps) => (StatusCode::OK, Json(caps)).into_response(),
        Err(e) => broker_error_response(e),
    }
}

/// Map a broker error to an HTTP response. Messages are secret-free.
fn broker_error_response(e: auth::BrokerError) -> axum::response::Response {
    let status = match e {
        auth::BrokerError::UnknownProvider(_)
        | auth::BrokerError::Denied(_)
        | auth::BrokerError::InvalidAccount(_)
        | auth::BrokerError::UnsupportedAccount { .. }
        | auth::BrokerError::UnsupportedCapability { .. } => StatusCode::BAD_REQUEST,
        auth::BrokerError::NotConfigured(_) => StatusCode::FORBIDDEN,
        auth::BrokerError::UnknownAccount { .. } => StatusCode::NOT_FOUND,
        auth::BrokerError::NoAccountAvailable { .. } => StatusCode::SERVICE_UNAVAILABLE,
        auth::BrokerError::Unauthorized => StatusCode::UNAUTHORIZED,
        _ => StatusCode::BAD_GATEWAY,
    };
    (status, Json(json!({ "error": e.to_string() }))).into_response()
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use futures::stream;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    // ── ct_eq (preserved) ────────────────────────────────────────────────────

    #[test]
    fn ct_eq_matches_only_identical() {
        assert!(ct_eq(b"Bearer secret", b"Bearer secret"));
        assert!(!ct_eq(b"Bearer secret", b"Bearer secreT"));
        assert!(!ct_eq(b"Bearer secret", b"Bearer wrong"));
        assert!(!ct_eq(b"short", b"longer-value"));
        assert!(!ct_eq(b"", b"x"));
        assert!(ct_eq(b"", b""));
    }

    // ── Bind-policy matrix (pure fn, no I/O) ─────────────────────────────────

    #[test]
    fn policy_loopback_no_tls_is_plain_http() {
        assert_eq!(
            check_bind_policy(true, false, false),
            BindDecision::Http { warn: false }
        );
    }

    #[test]
    fn policy_loopback_insecure_flag_is_still_plain_http_no_warn() {
        // loopback + insecure flag → still HTTP, no warning (flag is irrelevant on loopback)
        assert_eq!(
            check_bind_policy(true, false, true),
            BindDecision::Http { warn: false }
        );
    }

    #[test]
    fn policy_loopback_with_tls_is_https() {
        assert_eq!(check_bind_policy(true, true, false), BindDecision::Tls);
    }

    #[test]
    fn policy_nonloopback_no_tls_no_flag_refuses() {
        let d = check_bind_policy(false, false, false);
        assert!(matches!(d, BindDecision::Refuse(_)));
        if let BindDecision::Refuse(msg) = d {
            assert!(
                msg.contains("non-loopback"),
                "message should mention non-loopback: {msg}"
            );
            assert!(
                msg.contains("--tls-cert"),
                "message should mention --tls-cert: {msg}"
            );
            assert!(
                msg.contains("--insecure-http"),
                "message should mention --insecure-http: {msg}"
            );
        }
    }

    #[test]
    fn policy_nonloopback_no_tls_insecure_flag_warns() {
        assert_eq!(
            check_bind_policy(false, false, true),
            BindDecision::Http { warn: true }
        );
    }

    #[test]
    fn policy_nonloopback_with_tls_is_https() {
        assert_eq!(check_bind_policy(false, true, false), BindDecision::Tls);
    }

    #[test]
    fn policy_nonloopback_tls_and_insecure_flag_still_tls() {
        // TLS wins — insecure-http is irrelevant when TLS is configured.
        assert_eq!(check_bind_policy(false, true, true), BindDecision::Tls);
    }

    // ── PEM validation unit tests ─────────────────────────────────────────────

    #[test]
    fn validate_garbage_pem_cert_returns_error() {
        let garbage = b"this is not a pem file at all";
        let valid_key = generate_test_cert_key().1;
        let err = validate_pem_pair(garbage, &valid_key);
        assert!(err.is_err(), "garbage cert PEM should fail validation");
        let msg = err.unwrap_err().to_string();
        assert!(
            msg.contains("cert") || msg.contains("no certificates"),
            "error should mention cert issue: {msg}"
        );
    }

    #[test]
    fn validate_garbage_pem_key_returns_error() {
        let (valid_cert, _) = generate_test_cert_key();
        let garbage_key = b"this is definitely not a private key";
        let err = validate_pem_pair(&valid_cert, garbage_key);
        assert!(err.is_err(), "garbage key PEM should fail validation");
        let msg = err.unwrap_err().to_string();
        assert!(
            msg.contains("key") || msg.contains("no private key"),
            "error should mention key issue: {msg}"
        );
    }

    #[test]
    fn validate_valid_self_signed_pair_succeeds() {
        let (cert_pem, key_pem) = generate_test_cert_key();
        validate_pem_pair(&cert_pem, &key_pem).expect("self-signed pair should validate");
    }

    #[test]
    fn validate_cert_pem_used_as_key_returns_error() {
        // Providing cert PEM where key is expected → should fail.
        let (cert_pem, _) = generate_test_cert_key();
        let err = validate_pem_pair(&cert_pem, &cert_pem);
        assert!(err.is_err(), "cert-as-key should fail validation");
    }

    // ── load_and_validate_tls file-path tests ─────────────────────────────────

    #[test]
    fn load_tls_missing_cert_file_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let cert_path = dir.path().join("nonexistent-cert.pem");
        let key_path = dir.path().join("nonexistent-key.pem");
        // Key exists but cert doesn't
        let (_, key_pem) = generate_test_cert_key();
        std::fs::write(&key_path, &key_pem).unwrap();
        let err = load_and_validate_tls(&cert_path, &key_path);
        assert!(err.is_err());
        let msg = err.unwrap_err().to_string();
        assert!(
            msg.contains("--tls-cert"),
            "error should mention --tls-cert flag: {msg}"
        );
    }

    #[test]
    fn load_tls_missing_key_file_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("nonexistent-key.pem");
        let (cert_pem, _) = generate_test_cert_key();
        std::fs::write(&cert_path, &cert_pem).unwrap();
        let err = load_and_validate_tls(&cert_path, &key_path);
        assert!(err.is_err());
        let msg = err.unwrap_err().to_string();
        assert!(
            msg.contains("--tls-key"),
            "error should mention --tls-key flag: {msg}"
        );
    }

    #[test]
    fn load_tls_garbage_files_returns_error() {
        let dir = tempfile::tempdir().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");
        std::fs::write(&cert_path, b"not a cert").unwrap();
        std::fs::write(&key_path, b"not a key").unwrap();
        let err = load_and_validate_tls(&cert_path, &key_path);
        assert!(err.is_err(), "garbage files should fail validation");
    }

    #[test]
    fn load_tls_valid_pair_from_tempdir() {
        let dir = tempfile::tempdir().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");
        let (cert_pem, key_pem) = generate_test_cert_key();
        std::fs::write(&cert_path, &cert_pem).unwrap();
        std::fs::write(&key_path, &key_pem).unwrap();
        load_and_validate_tls(&cert_path, &key_path).expect("valid pair from files should succeed");
    }

    // ── TLS integration tests (axum-server + reqwest over real TLS) ───────────
    //
    // Spin up a real HTTPS listener on 127.0.0.1:<ephemeral>, hit it with
    // reqwest (danger_accept_invalid_certs for the self-signed test CA),
    // then gracefully shut down.

    /// Build a minimal HTTPS router for integration tests (no auth::load_auth dependency).
    fn build_test_router(machine_token: Option<String>) -> Router {
        let tok = machine_token.clone();
        Router::new()
            .route(
                "/healthz",
                get(|| async { Json(json!({ "status": "ok" })) }),
            )
            .route(
                "/token",
                get(move |headers: HeaderMap| {
                    let tok = tok.clone();
                    async move {
                        if let Some(ref expected) = tok {
                            let got = headers
                                .get("authorization")
                                .and_then(|v| v.to_str().ok())
                                .unwrap_or("");
                            let want = format!("Bearer {expected}");
                            if !ct_eq(got.as_bytes(), want.as_bytes()) {
                                return (
                                    StatusCode::UNAUTHORIZED,
                                    Json(json!({ "error": "bad machine auth" })),
                                );
                            }
                        }
                        (
                            StatusCode::OK,
                            Json(json!({
                                "access_token": "test-tok",
                                "expires": 9_999_999_999u64,
                                "ttl_ms": 999_999u64
                            })),
                        )
                    }
                }),
            )
    }

    /// Spin up a real TLS listener on a random loopback port; return addr + handle.
    async fn spawn_tls_server(
        machine_token: Option<String>,
    ) -> (SocketAddr, axum_server::Handle<SocketAddr>) {
        // rustls 0.23 requires an explicit CryptoProvider when multiple backends
        // are compiled in (both aws-lc-rs and ring are in the dep tree here).
        // install_default() is idempotent — the Err case just means it was already set.
        let _ = rustls::crypto::ring::default_provider().install_default();

        let (cert_pem, key_pem) = generate_test_cert_key();
        let rustls_config = axum_server::tls_rustls::RustlsConfig::from_pem(cert_pem, key_pem)
            .await
            .expect("test TLS config should build");

        let app = build_test_router(machine_token);

        // Port 0 → let the OS assign an ephemeral port. Keep the listener alive
        // and pass it directly to from_tcp_rustls so there's no TOCTOU window
        // between get-addr and re-bind.
        let std_listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        std_listener.set_nonblocking(true).unwrap();
        let addr = std_listener.local_addr().unwrap();

        let handle = axum_server::Handle::<SocketAddr>::new();
        let h2 = handle.clone();
        tokio::spawn(async move {
            axum_server::tls_rustls::from_tcp_rustls(std_listener, rustls_config)
                .expect("from_tcp_rustls should succeed")
                .handle(h2)
                .serve(app.into_make_service())
                .await
                .ok();
        });
        // Give it a tick to bind and start accepting.
        tokio::time::sleep(Duration::from_millis(80)).await;
        (addr, handle)
    }

    /// reqwest client that accepts self-signed certs (test-only, never ship to prod).
    fn insecure_client() -> reqwest::Client {
        reqwest::Client::builder()
            .danger_accept_invalid_certs(true)
            .build()
            .expect("insecure reqwest client should build")
    }

    #[tokio::test]
    async fn tls_healthz_returns_200() {
        let (addr, handle) = spawn_tls_server(None).await;
        let url = format!("https://{addr}/healthz");
        let resp = insecure_client()
            .get(&url)
            .send()
            .await
            .expect("GET /healthz over TLS");
        assert_eq!(resp.status(), 200, "/healthz should return 200 over TLS");
        let body: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(body["status"], "ok");
        handle.graceful_shutdown(Some(Duration::from_millis(200)));
    }

    #[tokio::test]
    async fn tls_token_no_auth_returns_401() {
        let (addr, handle) = spawn_tls_server(Some("secret123".to_string())).await;
        let url = format!("https://{addr}/token");
        let resp = insecure_client()
            .get(&url)
            .send()
            .await
            .expect("GET /token no auth");
        assert_eq!(
            resp.status(),
            401,
            "/token without bearer should 401 over TLS"
        );
        handle.graceful_shutdown(Some(Duration::from_millis(200)));
    }

    #[tokio::test]
    async fn tls_token_with_correct_auth_returns_200() {
        let (addr, handle) = spawn_tls_server(Some("my-machine-token".to_string())).await;
        let url = format!("https://{addr}/token");
        let resp = insecure_client()
            .get(&url)
            .header("Authorization", "Bearer my-machine-token")
            .send()
            .await
            .expect("GET /token with correct auth");
        assert_eq!(
            resp.status(),
            200,
            "/token with correct bearer should 200 over TLS"
        );
        let body: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(body["access_token"], "test-tok");
        handle.graceful_shutdown(Some(Duration::from_millis(200)));
    }

    #[tokio::test]
    async fn tls_token_wrong_auth_returns_401() {
        let (addr, handle) = spawn_tls_server(Some("my-machine-token".to_string())).await;
        let url = format!("https://{addr}/token");
        let resp = insecure_client()
            .get(&url)
            .header("Authorization", "Bearer wrong-token")
            .send()
            .await
            .expect("GET /token with wrong auth");
        assert_eq!(
            resp.status(),
            401,
            "/token with wrong bearer should 401 over TLS"
        );
        handle.graceful_shutdown(Some(Duration::from_millis(200)));
    }

    // ── Broker service: /proxy, /capabilities, /token isolation ─────────────

    // ── Genuine authenticated cloud production-router E2E ────────────────────

    struct TypedCloudFixture {
        invokes: AtomicUsize,
    }

    #[async_trait]
    impl auth::broker::CloudBackend for TypedCloudFixture {
        async fn catalog(
            &self,
            provider: auth::CloudProviderId,
            context: &str,
            _: bool,
        ) -> Result<Vec<auth::broker::CloudCatalogEntry>, auth::BrokerError> {
            assert_eq!(context, "ctx-opaque-route");
            Ok(vec![auth::broker::CloudCatalogEntry {
                provider,
                id: "model-x".into(),
                display_name: "Model X".into(),
                context_ref: context.into(),
                context_label: "sandbox".into(),
                stale: false,
                fetched_at: 7,
            }])
        }
        async fn invoke(
            &self,
            _: auth::CloudProviderId,
            context: &str,
            model: &str,
            _: auth::InvokeRequest,
        ) -> Result<auth::broker::CloudEventStream, auth::BrokerError> {
            assert_eq!((context, model), ("ctx-opaque-route", "model-x"));
            self.invokes.fetch_add(1, Ordering::SeqCst);
            Ok(Box::pin(stream::iter(vec![
                Ok(auth::broker::CloudEvent::TextDelta {
                    delta: "frag".into(),
                }),
                Ok(auth::broker::CloudEvent::TextDelta {
                    delta: "mented".into(),
                }),
                Ok(auth::broker::CloudEvent::Done),
            ])))
        }
    }

    #[tokio::test]
    async fn remote_broker_catalog_then_fragmented_invoke_uses_real_auth_router() {
        use auth::CredentialBroker;
        let backend = Arc::new(TypedCloudFixture {
            invokes: AtomicUsize::new(0),
        });
        let app = build_router_with_cloud_backend(Some("machine-only".into()), backend.clone());
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(
                listener,
                app.into_make_service_with_connect_info::<SocketAddr>(),
            )
            .await
            .unwrap()
        });
        let endpoint = format!("http://{addr}");
        let remote = auth::RemoteBroker::new(
            &endpoint,
            "machine-only",
            reqwest::Client::new(),
            auth::TokenCache::new(),
        );
        let provider = auth::CloudProviderId::AwsBedrock;
        let catalog = remote
            .cloud_catalog(provider, "ctx-opaque-route", false)
            .await
            .unwrap();
        assert_eq!(catalog[0].context_ref, "ctx-opaque-route");
        let request: auth::InvokeRequest =
            serde_json::from_value(json!({"messages":[],"tools":[],"stream":true,"options":{}}))
                .unwrap();
        let mut events = remote
            .cloud_invoke(
                provider,
                &catalog[0].context_ref,
                &catalog[0].id,
                request.clone(),
            )
            .await
            .unwrap();
        let mut seen = Vec::new();
        while let Some(event) = events.next().await {
            seen.push(event.unwrap());
        }
        assert_eq!(
            seen,
            vec![
                auth::broker::CloudEvent::TextDelta {
                    delta: "frag".into()
                },
                auth::broker::CloudEvent::TextDelta {
                    delta: "mented".into()
                },
                auth::broker::CloudEvent::Done
            ]
        );
        assert_eq!(backend.invokes.load(Ordering::SeqCst), 1);

        for token in [None, Some("invalid")] {
            let client = reqwest::Client::new();
            let mut call = client.post(format!("{endpoint}/cloud/invoke")).json(&json!({"provider":"aws-bedrock","context_ref":"ctx-opaque-route","model_id":"model-x","request":request}));
            if let Some(token) = token {
                call = call.bearer_auth(token);
            }
            let response = call.send().await.unwrap();
            assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
            let wire = response.text().await.unwrap();
            assert!(!wire.contains("machine-only") && !wire.contains("ctx-opaque-route"));
        }
        assert_eq!(backend.invokes.load(Ordering::SeqCst), 1);
    }

    /// Spawn the REAL router (real handlers/state) on an ephemeral loopback
    /// port. `local_upstream` pins the `local` provider at a fake endpoint so
    /// no real provider or credential file is touched.
    async fn spawn_broker_service(
        machine_token: Option<String>,
        local_upstream: Option<String>,
    ) -> String {
        spawn_broker_service_with(machine_token, local_upstream, true).await
    }

    /// `loopback_host_check`: run the production loopback Host allowlist
    /// for the bound port (what `run` installs for a loopback bind).
    async fn spawn_broker_service_with(
        machine_token: Option<String>,
        local_upstream: Option<String>,
        loopback_host_check: bool,
    ) -> String {
        let client = reqwest::Client::new();
        let local = match local_upstream {
            Some(url) => auth::LocalBroker::with_local_base_url(client.clone(), url),
            None => auth::LocalBroker::new(client.clone()),
        };
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let state = BrokerState {
            machine_token,
            allowed_hosts: loopback_host_check
                .then(|| Arc::from(loopback_allowed_hostnames(addr.ip()))),
            token_source: Arc::from("test"),
            local: Arc::new(local),
        };
        let app = build_router(state);
        tokio::spawn(async move {
            axum::serve(
                listener,
                app.into_make_service_with_connect_info::<SocketAddr>(),
            )
            .await
            .unwrap();
        });
        format!("http://{addr}")
    }

    /// Fake OpenAI-compatible upstream that records the Authorization header
    /// it receives and serves an SSE body.
    async fn spawn_sse_upstream(seen_auth: Arc<std::sync::Mutex<String>>) -> String {
        use axum::routing::post as axum_post;
        let app = Router::new().route(
            "/chat/completions",
            axum_post(move |headers: HeaderMap| {
                let seen = seen_auth.clone();
                async move {
                    *seen.lock().unwrap() = headers
                        .get("authorization")
                        .and_then(|v| v.to_str().ok())
                        .unwrap_or("")
                        .to_string();
                    (
                        [("content-type", "text/event-stream")],
                        "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\ndata: [DONE]\n\n",
                    )
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        format!("http://{addr}")
    }

    fn proxy_body(stream: bool) -> serde_json::Value {
        json!({
            "provider": "local",
            "method": "post",
            "path": "/chat/completions",
            "body": {"model": "m", "messages": []},
            "stream": stream,
        })
    }

    /// Proxy authorization: no bearer and a wrong bearer are both 401, and the
    /// denial body contains no credential material.
    #[tokio::test]
    async fn proxy_requires_machine_auth() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();

        let resp = client
            .post(format!("{base}/proxy"))
            .json(&proxy_body(false))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 401, "missing bearer must be denied");

        let resp = client
            .post(format!("{base}/proxy"))
            .bearer_auth("wrong-token")
            .json(&proxy_body(false))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 401, "wrong bearer must be denied");
        let body = resp.text().await.unwrap();
        assert!(
            !body.contains("machine-secret"),
            "denial must not echo the expected token"
        );
    }

    /// The typed /usage operation is a credential endpoint: it requires
    /// machine auth, and denials carry no token material.
    #[tokio::test]
    async fn usage_requires_machine_auth_and_denials_are_secret_free() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();

        let resp = client.get(format!("{base}/usage")).send().await.unwrap();
        assert_eq!(resp.status(), 401, "missing bearer must be denied");

        let resp = client
            .get(format!("{base}/usage"))
            .bearer_auth("wrong-token")
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 401, "wrong bearer must be denied");
        let body = resp.text().await.unwrap();
        assert!(
            !body.contains("machine-secret") && !body.contains("access"),
            "denial must not echo credentials: {body}"
        );
    }

    /// Capabilities also require machine auth (they reveal configured-ness).
    #[tokio::test]
    async fn capabilities_require_machine_auth_and_expose_no_secrets() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();

        let resp = client
            .get(format!("{base}/capabilities"))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 401);

        let resp = client
            .get(format!("{base}/capabilities"))
            .bearer_auth("machine-secret")
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 200);
        let caps: serde_json::Value = resp.json().await.unwrap();
        for row in caps.as_array().expect("array") {
            let obj = row.as_object().unwrap();
            for field in obj.keys() {
                assert!(
                    matches!(
                        field.as_str(),
                        "configured" | "key" | "kind" | "name" | "accounts"
                    ),
                    "capability rows carry status only — unexpected field {field}"
                );
            }
            assert!(obj["configured"].is_boolean());
            // Account rows (OAuth providers only) are non-secret listing rows.
            if let Some(accounts) = obj.get("accounts").and_then(|a| a.as_array()) {
                for account in accounts {
                    for field in account.as_object().unwrap().keys() {
                        assert!(
                            matches!(
                                field.as_str(),
                                "provider"
                                    | "label"
                                    | "identity"
                                    | "account_id_prefix"
                                    // Opaque SHA-256 over the provider account id
                                    // (never token material); lets a client pin
                                    // spending to exactly one listed seat.
                                    | "seat_fingerprint"
                                    | "expires"
                                    | "added_at"
                                    | "selected"
                                    | "cooldown_until"
                            ),
                            "account rows must never carry credential fields: {field}"
                        );
                    }
                    let prefix = account
                        .get("account_id_prefix")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");
                    assert!(prefix.len() <= 8, "account id must be a short prefix");
                }
            }
        }
    }

    /// Remote non-disclosure / cross-provider isolation: the /token endpoint
    /// refuses to vend anything for a static-key provider — static keys are
    /// proxy-only and never leave the broker.
    #[tokio::test]
    async fn token_endpoint_denies_static_key_providers() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();
        for provider in ["groq", "openrouter", "local", "definitely-unknown"] {
            let resp = client
                .get(format!("{base}/token"))
                .query(&[("provider", provider)])
                .bearer_auth("machine-secret")
                .send()
                .await
                .unwrap();
            assert_eq!(
                resp.status(),
                400,
                "static/unknown provider '{provider}' must never be vended a token"
            );
            let body = resp.text().await.unwrap();
            assert!(
                !body.contains("access_token"),
                "no token material for {provider}"
            );
        }
    }

    // ── Loopback Host allowlist (DNS rebinding, #147 C1) ─────────────────────

    async fn capabilities_with_host(base: &str, host: &str) -> u16 {
        reqwest::Client::new()
            .get(format!("{base}/capabilities"))
            .header(reqwest::header::HOST, host)
            .bearer_auth("machine-secret")
            .send()
            .await
            .unwrap()
            .status()
            .as_u16()
    }

    /// prA I4: the hostname is checked, the port is not (DNS rebinding sends
    /// a foreign hostname; an `ssh -L 9000:127.0.0.1:<port>` forward sends
    /// `Host: 127.0.0.1:9000` and must work).
    #[tokio::test]
    async fn loopback_broker_rejects_foreign_hostnames_but_not_other_ports() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let port = base.rsplit(':').next().unwrap().to_string();
        for host in [
            format!("127.0.0.1:{port}"),
            format!("localhost:{port}"),
            format!("LOCALHOST:{port}"),
            format!("[::1]:{port}"),
            "127.0.0.1:9000".to_string(),
            "localhost:1".to_string(),
            "localhost".to_string(),
            "[::1]".to_string(),
        ] {
            assert_eq!(capabilities_with_host(&base, &host).await, 200, "{host}");
        }
        for host in [
            format!("evil.example:{port}"),
            "evil.example".to_string(),
            format!("localhost.evil.example:{port}"),
            format!("127.0.0.1.evil.example:{port}"),
            format!("user@127.0.0.1:{port}"),
            "127.0.0.2".to_string(),
            String::new(),
        ] {
            assert_eq!(
                capabilities_with_host(&base, &host).await,
                403,
                "rebinding Host {host:?} must be refused even with a valid token"
            );
        }
        // Every route is covered, including the unauthenticated /healthz.
        let resp = reqwest::Client::new()
            .get(format!("{base}/healthz"))
            .header(reqwest::header::HOST, "evil.example")
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 403);
        let body = resp.text().await.unwrap();
        assert!(!body.contains("machine-secret"));
    }

    /// An HTTP/1.0 request with no Host header at all is refused.
    #[tokio::test]
    async fn loopback_broker_refuses_a_missing_host_header() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let addr = base.trim_start_matches("http://");
        let mut conn = tokio::net::TcpStream::connect(addr).await.unwrap();
        conn.write_all(b"GET /healthz HTTP/1.0\r\n\r\n").await.unwrap();
        let mut reply = String::new();
        conn.read_to_string(&mut reply).await.unwrap();
        assert!(
            reply.starts_with("HTTP/1.0 403") || reply.starts_with("HTTP/1.1 403"),
            "{reply}"
        );
    }

    /// prA m6: a broker bound to another 127.x address accepts its own IP.
    #[tokio::test]
    async fn loopback_broker_on_another_127_address_accepts_its_bind_ip() {
        let listener = match tokio::net::TcpListener::bind("127.0.0.2:0").await {
            Ok(l) => l,
            Err(_) => return, // platform without 127/8 aliasing (macOS)
        };
        let addr = listener.local_addr().unwrap();
        let state = BrokerState {
            machine_token: Some("machine-secret".into()),
            allowed_hosts: Some(Arc::from(loopback_allowed_hostnames(addr.ip()))),
            token_source: Arc::from("test"),
            local: Arc::new(auth::LocalBroker::new(reqwest::Client::new())),
        };
        let app = build_router(state);
        tokio::spawn(async move {
            axum::serve(
                listener,
                app.into_make_service_with_connect_info::<SocketAddr>(),
            )
            .await
            .unwrap();
        });
        let base = format!("http://{addr}");
        assert_eq!(capabilities_with_host(&base, &addr.to_string()).await, 200);
        assert_eq!(capabilities_with_host(&base, "127.0.0.3:1").await, 403);
    }

    #[tokio::test]
    async fn host_allowlist_is_only_applied_to_loopback_binds() {
        let base = spawn_broker_service_with(Some("machine-secret".into()), None, false).await;
        assert_eq!(capabilities_with_host(&base, "broker.example:8181").await, 200);
    }

    #[test]
    fn loopback_allowed_hostnames_cover_the_loopback_names_and_bind_ip() {
        let v4: std::net::IpAddr = "127.0.0.1".parse().unwrap();
        assert_eq!(
            loopback_allowed_hostnames(v4),
            vec!["localhost", "127.0.0.1", "[::1]"]
        );
        let other: std::net::IpAddr = "127.0.0.2".parse().unwrap();
        assert!(loopback_allowed_hostnames(other).contains(&"127.0.0.2".to_string()));
        assert_eq!(host_header_hostname("[::1]:8181").as_deref(), Some("[::1]"));
        assert_eq!(host_header_hostname("LOCALHOST").as_deref(), Some("LOCALHOST"));
        assert_eq!(host_header_hostname(""), None);
    }

    #[test]
    fn denial_key_is_per_reason_route_and_peer() {
        let a: std::net::IpAddr = "127.0.0.1".parse().unwrap();
        let b: std::net::IpAddr = "10.0.0.2".parse().unwrap();
        let k = denial_key(false, "/usage", Some(a));
        assert_eq!(k, "missing|/usage|127.0.0.1");
        assert_ne!(k, denial_key(false, "/capabilities", Some(a)), "per route");
        assert_ne!(k, denial_key(false, "/usage", Some(b)), "per peer");
        assert_ne!(k, denial_key(true, "/usage", Some(a)), "per reason");
    }

    #[test]
    fn denial_advisory_is_rate_limited_per_reason_and_peer() {
        let log = DenialLog::default();
        let t0 = std::time::Instant::now();
        let every = Duration::from_secs(60);
        assert_eq!(denial_advisory_due(&log, "missing|127.0.0.1", t0, every), Some(0));
        for i in 1..=50 {
            let t = t0 + Duration::from_millis(i * 100);
            assert_eq!(denial_advisory_due(&log, "missing|127.0.0.1", t, every), None);
        }
        // A different reason or peer is its own stream.
        assert_eq!(denial_advisory_due(&log, "wrong|127.0.0.1", t0, every), Some(0));
        assert_eq!(denial_advisory_due(&log, "missing|10.0.0.2", t0, every), Some(0));
        // After the interval: logged again, with the suppressed count.
        let later = t0 + Duration::from_secs(61);
        assert_eq!(denial_advisory_due(&log, "missing|127.0.0.1", later, every), Some(50));
        assert_eq!(denial_advisory_due(&log, "missing|127.0.0.1", later, every), None);
    }

    /// prA I2: a denied request gets an actionable hint and never the token.
    #[tokio::test]
    async fn machine_auth_denial_carries_a_hint_and_no_token() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let resp = reqwest::Client::new()
            .get(format!("{base}/capabilities"))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 401);
        let body: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(body["error"], "bad machine auth");
        let hint = body["hint"].as_str().unwrap();
        assert!(
            hint.contains("auth.machine_token") && hint.contains("SYNAPS_MACHINE_TOKEN"),
            "{hint}"
        );
        assert!(hint.contains("auth-broker.token") && hint.contains("restart"), "{hint}");
        assert!(!body.to_string().contains("machine-secret"));
    }

    // ── Loopback machine-token bootstrap (#147 C1) ───────────────────────────

    #[test]
    fn loopback_without_token_generates_one_and_remote_needs_explicit_opt_out() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("auth-broker.token");
        let tok = resolve_machine_token(None, true, false, &path).unwrap();
        let tok = tok.expect("loopback always has a machine token");
        assert_eq!(std::fs::read_to_string(&path).unwrap().trim(), tok);
        assert_eq!(
            resolve_machine_token(None, true, false, &path).unwrap().as_deref(),
            Some(tok.as_str()),
            "restart reuses the token clients already read"
        );
        // --insecure-no-auth never turns auth off on loopback.
        assert_eq!(
            resolve_machine_token(None, true, true, &path).unwrap().as_deref(),
            Some(tok.as_str())
        );
        // An explicit token wins and no file is consulted.
        let other = dir.path().join("unused.token");
        assert_eq!(
            resolve_machine_token(Some("explicit".into()), true, false, &other)
                .unwrap()
                .as_deref(),
            Some("explicit")
        );
        assert!(!other.exists());
        // Non-loopback: unchanged — no token is refused unless opted out.
        assert!(resolve_machine_token(None, false, false, &other).is_err());
        assert_eq!(resolve_machine_token(None, false, true, &other).unwrap(), None);
        assert!(!other.exists(), "never auto-generated for a network bind");
    }

    // ── /usage/snapshot and /accounts/cooldown machine auth (#147 I5) ───────
    //
    // With the right bearer the request reaches the handler's own validation
    // (400 for a bad provider / account), which needs no credential store;
    // without it, or with a wrong one, it is 401 before anything else runs.

    async fn snapshot_status(base: &str, bearer: Option<&str>, query: &[(&str, &str)]) -> (u16, String) {
        let mut req = reqwest::Client::new()
            .get(format!("{base}/usage/snapshot"))
            .query(query);
        if let Some(b) = bearer {
            req = req.bearer_auth(b);
        }
        let resp = req.send().await.unwrap();
        (resp.status().as_u16(), resp.text().await.unwrap())
    }

    async fn cooldown_status(base: &str, bearer: Option<&str>, body: serde_json::Value) -> (u16, String) {
        let mut req = reqwest::Client::new()
            .post(format!("{base}/accounts/cooldown"))
            .json(&body);
        if let Some(b) = bearer {
            req = req.bearer_auth(b);
        }
        let resp = req.send().await.unwrap();
        (resp.status().as_u16(), resp.text().await.unwrap())
    }

    #[tokio::test]
    async fn usage_snapshot_requires_machine_auth() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let q = [("provider", "anthropic"), ("account", "../bad")];
        for bearer in [None, Some("wrong-token"), Some("machine-secret-but-longer")] {
            let (status, body) = snapshot_status(&base, bearer, &q).await;
            assert_eq!(status, 401, "{bearer:?}");
            assert!(!body.contains("machine-secret"), "{body}");
        }
        let (status, body) = snapshot_status(&base, Some("machine-secret"), &q).await;
        assert_eq!(status, 400, "authorized request reaches validation: {body}");
        assert!(body.contains("invalid account"), "{body}");
        let (status, _) = snapshot_status(
            &base,
            Some("machine-secret"),
            &[("provider", "groq"), ("account", "default")],
        )
        .await;
        assert_eq!(status, 400, "static-key provider is not a snapshot target");
    }

    #[tokio::test]
    async fn accounts_cooldown_requires_machine_auth() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let body = json!({"provider": "anthropic", "account": "../bad", "reason": "t"});
        for bearer in [None, Some("wrong-token")] {
            let (status, text) = cooldown_status(&base, bearer, body.clone()).await;
            assert_eq!(status, 401, "{bearer:?}");
            assert!(!text.contains("machine-secret"), "{text}");
        }
        let (status, text) = cooldown_status(&base, Some("machine-secret"), body).await;
        assert_eq!(status, 400, "authorized request reaches validation: {text}");
        assert!(text.contains("invalid account"), "{text}");
        let (status, text) = cooldown_status(
            &base,
            Some("machine-secret"),
            json!({"provider": "groq", "account": "default"}),
        )
        .await;
        assert_eq!(status, 400, "{text}");
        assert!(text.contains("unknown provider"), "{text}");
    }

    /// Malformed `account` on `/token` is 400 from the real handler before
    /// any credential lookup (unknown → 404 and auto → 503 need a store;
    /// see tests/multi_account_cli.rs, which runs the production binary).
    #[tokio::test]
    async fn token_malformed_account_is_400() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        for account in ["../bad", "Bad Label", "", "a/b"] {
            let resp = reqwest::Client::new()
                .get(format!("{base}/token"))
                .query(&[("provider", "anthropic"), ("account", account)])
                .bearer_auth("machine-secret")
                .send()
                .await
                .unwrap();
            assert_eq!(resp.status(), 400, "{account:?}");
            let body = resp.text().await.unwrap();
            assert!(body.contains("invalid account") && !body.contains("access_token"), "{body}");
        }
    }

    // ── Muse is proxy-only (#216 I2) ────────────────────────────────────────

    /// The minted Muse Model API key never expires on its own and Muse is
    /// routed BrokerProxy, so `/token` must never vend it — for any slot.
    #[tokio::test]
    async fn token_endpoint_never_vends_muse() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();
        for query in [
            vec![("provider", "muse")],
            vec![("provider", "muse"), ("account", "default")],
            vec![("provider", "muse"), ("account", "work")],
            vec![("provider", "muse"), ("account", "auto")],
        ] {
            let resp = client
                .get(format!("{base}/token"))
                .query(&query)
                .bearer_auth("machine-secret")
                .send()
                .await
                .unwrap();
            assert_eq!(resp.status(), 400, "{query:?}");
            let body = resp.text().await.unwrap();
            assert!(!body.contains("access_token"), "{query:?}: {body}");
        }
    }

    /// `/proxy` is unaffected: a Muse proxy request still reaches the Muse
    /// endpoint policy (here: its allowlist refusing the key-mint path,
    /// checked before any credential or network access).
    #[tokio::test]
    async fn proxy_still_routes_muse_requests() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let resp = reqwest::Client::new()
            .post(format!("{base}/proxy"))
            .bearer_auth("machine-secret")
            .json(&json!({
                "provider": "muse",
                "method": "post",
                "path": "/muse-code/key",
                "body": {},
                "stream": false,
            }))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 400);
        let body = resp.text().await.unwrap();
        assert!(
            !body.contains("unknown provider"),
            "muse is still a proxy provider: {body}"
        );
        assert!(body.contains("allow"), "the Muse path allowlist answered: {body}");
    }

    #[test]
    fn muse_is_excluded_from_token_vending_only() {
        assert_eq!(token_vending_provider("muse"), None);
        // Usage snapshots / cooldown reports keep their provider check.
        assert_eq!(broker_provider("muse"), Some(auth::OAuthProviderId::Muse));
        assert_eq!(
            token_vending_provider("anthropic"),
            Some(auth::OAuthProviderId::Anthropic)
        );
    }

    #[test]
    fn github_copilot_is_oauth_token_vending_allowlisted() {
        // Descriptor strategy permits /token for github-copilot (session token only).
        let id = broker_provider("github-copilot").expect("github-copilot must be allowlisted");
        assert_eq!(id, auth::OAuthProviderId::GitHubCopilot);
        assert_eq!(id.as_str(), "github-copilot");
        // Aliases are not canonical broker ids
        assert!(broker_provider("copilot").is_none());
        assert!(broker_provider("gh-copilot").is_none());
    }

    /// Proxy validation fails closed: OAuth providers and absolute/escaping
    /// paths are rejected before any upstream contact.
    #[tokio::test]
    async fn proxy_rejects_oauth_providers_and_bad_paths() {
        let base = spawn_broker_service(Some("machine-secret".into()), None).await;
        let client = reqwest::Client::new();

        for (provider, path) in [
            ("anthropic", "/v1/messages"),
            ("openai-codex", "/responses"),
            // `/v1/messages` is advertised by Anthropic-vendor Copilot rows but
            // the broker does not currently route it — must fail closed.
            ("github-copilot", "/v1/messages"),
            ("github-copilot", "/models/gpt-5.4/policy"),
            ("local", "https://evil.example/steal"),
            ("local", "/../../etc/passwd"),
        ] {
            let resp = client
                .post(format!("{base}/proxy"))
                .bearer_auth("machine-secret")
                .json(&json!({
                    "provider": provider,
                    "method": "post",
                    "path": path,
                    "stream": false,
                }))
                .send()
                .await
                .unwrap();
            assert_eq!(resp.status(), 400, "{provider} {path} must be rejected");
        }
    }

    /// Streaming forwarding: an authenticated /proxy stream call reaches the
    /// upstream with the BROKER-applied credential and the SSE bytes flow back
    /// to the client unmodified. The client never supplied a provider key.
    #[tokio::test]
    async fn proxy_streams_sse_and_applies_key_broker_side() {
        let seen_auth = Arc::new(std::sync::Mutex::new(String::new()));
        let upstream = spawn_sse_upstream(seen_auth.clone()).await;
        let base = spawn_broker_service(Some("machine-secret".into()), Some(upstream)).await;
        let client = reqwest::Client::new();

        let resp = client
            .post(format!("{base}/proxy"))
            .bearer_auth("machine-secret")
            .json(&proxy_body(true))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 200);
        assert_eq!(
            resp.headers()
                .get("content-type")
                .and_then(|v| v.to_str().ok()),
            Some("text/event-stream")
        );
        let body = resp.text().await.unwrap();
        assert!(
            body.contains("data: {\"choices\""),
            "SSE payload forwarded: {body}"
        );
        assert!(body.contains("[DONE]"));
        // The upstream saw the broker-owned credential, not the machine token.
        assert_eq!(&*seen_auth.lock().unwrap(), "Bearer local");
    }

    /// Non-streaming proxy returns the JSON envelope RemoteBroker expects.
    #[tokio::test]
    async fn proxy_non_streaming_returns_status_envelope() {
        let seen_auth = Arc::new(std::sync::Mutex::new(String::new()));
        let upstream = spawn_sse_upstream(seen_auth.clone()).await;
        let base = spawn_broker_service(None, Some(upstream)).await;
        let client = reqwest::Client::new();
        let resp = client
            .post(format!("{base}/proxy"))
            .json(&proxy_body(false))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 200);
        let envelope: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(envelope["status"], 200);
        assert!(envelope["body"].as_str().unwrap().contains("[DONE]"));
    }

    // ── Helper: rcgen self-signed cert ────────────────────────────────────────

    /// Mint a self-signed cert + PKCS#8 key pair for localhost.
    /// Returns (cert_pem_bytes, key_pem_bytes).
    pub(super) fn generate_test_cert_key() -> (Vec<u8>, Vec<u8>) {
        use rcgen::generate_simple_self_signed;
        let san = vec!["localhost".to_string(), "127.0.0.1".to_string()];
        let cert = generate_simple_self_signed(san).expect("rcgen cert generation should succeed");
        let cert_pem = cert.cert.pem().into_bytes();
        let key_pem = cert.signing_key.serialize_pem().into_bytes();
        (cert_pem, key_pem)
    }
}