holger-server-lib 0.6.9

//! HTTP/OCI gateway: the one non-gRPC ingress, letting HTTP-only clients (helm,
//! pip, cargo, OCI registry pushes via `oci://…`) reach the same repository
//! backends the tonic service in `grpc.rs` serves — each request is routed to a
//! backend and dispatched through its `handle_http2_request`. [`GatewaySettings`]
//! carries the shared [`FastRoutes`], auth config, optional TLS, body cap, and
//! the audit sink into every connection.
//!
//! Security-app rules, deliberately kept in lockstep with the gRPC path:
//! TLS terminates here and mTLS CN is lifted from the handshake once per
//! connection; reads (GET/HEAD) stay open and audit as `anonymous`, while writes
//! must pass `auth::validate_request` then `authorize_http_write` (RBAC mirror of
//! `grpc::authorize_write`, DELETE = admin, other writes = writer-or-admin,
//! fail-closed under a configured policy). Only `/healthz`, `/readyz`, and the
//! `/v2` OCI version probe are unauthenticated and unaudited. Routing keys off
//! the first path segment (`/v2/{name}/…` for OCI, else `/{repo}/…`) via
//! [`route_key`]; backends strip their own prefix from the passed-through path.
//!
//! Gotcha: the read-only write-gate (`backend.is_writable()`) and the body-size
//! cap both fire BEFORE the body is collected, so a rejected write never buffers
//! the upload — this is intentional, closing the DoS surface the pre-gate HTTP
//! path left open (it used to hand writes to a read-only backend that silently
//! swallowed them).
//
// HTTP/OCI gateway: bridges HTTP clients (helm, pip, cargo, OCI registry) to
// the repository backends' `handle_http2_request` handlers. Helm and friends
// speak HTTP, not gRPC, so this listener is required for any ecosystem interop;
// the modern Helm flow (`helm push/pull oci://…`) rides the OCI Distribution
// API served here.
//
// Security posture (this is a security application):
//   * TLS is terminated here when the endpoint has `ron_tls`; client certs are
//     requested but optional, so anonymous pulls and Kubernetes probes work
//     while writes are gated on identity at the application layer.
//   * mTLS identity (client-cert CN) is extracted once per connection and fed
//     into auth alongside any Bearer token.
//   * Request bodies are size-capped to bound memory use (upload DoS guard).
//   * Reserved unauthenticated paths: GET /healthz, GET /readyz, GET /v2/.
//
// Routing: `/v2/{name}/…` → repo named by the first OCI name segment (OCI);
// `/{repo}/…` → repo named by the first path segment (classic). The full
// path+query is passed through; backends strip their own prefix.

use std::sync::Arc;

/// Per-repo listing cap for the `/metrics` artifact-count gauge. The gauge is an
/// operability signal, not an inventory, so a very large repo is counted up to
/// this bound rather than paying an unbounded scan on every scrape.
const METRICS_LIST_CAP: usize = 100_000;

use http_body_util::{BodyExt, Full, Limited};
use hyper::body::{Bytes, Incoming};
use hyper::service::service_fn;
use hyper::{Request, Response};
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as ConnBuilder;
use rustls::ServerConfig;
use tokio_rustls::TlsAcceptor;

use crate::audit::{AuditAction, AuditEvent, AuditLog};
use crate::auth::{self, AuthConfig};
use crate::exposed::fast_routes::FastRoutes;
use crate::exposed::tls::leaf_common_name;

/// Everything a gateway connection needs, shared behind an `Arc`.
pub struct GatewaySettings {
    pub routes: FastRoutes,
    pub auth_config: Arc<AuthConfig>,
    pub tls: Option<Arc<ServerConfig>>,
    pub max_body_bytes: usize,
    /// Append-only audit sink, shared with the gRPC services. Reads
    /// (GET/HEAD) record as anonymous downloads; gated writes record the
    /// resolved identity (Bearer sub / mTLS CN).
    pub audit: Arc<dyn AuditLog>,
    /// Hosted-SBOM store, shared with the `SbomService` gRPC RPCs. Backs the
    /// read-only `GET /-/sbom` door (attach is gRPC/CLI-only). `None` ⇒ the door
    /// answers 404 (SBOM hosting not configured on this server).
    pub sboms: Option<crate::sbom::SharedSbomStore>,
    /// Custom-property store, shared with the gRPC `Search` property axis. Backs
    /// the read-only `GET /-/properties` door AND the `property=` filter on
    /// `GET /-/search` (property WRITES are CLI/in-process for now). `None` ⇒ the
    /// door answers 404 and a `property=` filter matches nothing (fail-closed).
    pub properties: Option<crate::properties::SharedPropertyStore>,
    /// Serve-time **quarantine** predicate `(key, value)`, shared with the gRPC
    /// fetch gate. `Some` ⇒ a GET whose coordinate (via
    /// [`RepositoryBackendTrait::coordinate_for_path`]) has matching properties is
    /// refused with 404 before the backend serves it. `None` ⇒ no serve gate.
    pub quarantine: Option<(String, String)>,
}

pub fn start_http_gateway(
    addr: std::net::SocketAddr,
    settings: Arc<GatewaySettings>,
) -> anyhow::Result<()> {
    tokio::spawn(async move {
        let listener = match tokio::net::TcpListener::bind(addr).await {
            Ok(l) => l,
            Err(e) => {
                eprintln!("HTTP/OCI gateway failed to bind {}: {}", addr, e);
                return;
            }
        };
        let scheme = if settings.tls.is_some() { "https" } else { "http (cleartext)" };
        println!("HTTP/OCI gateway listening on {} [{}]", addr, scheme);
        if settings.tls.is_none() {
            log::warn!(
                "SECURITY: HTTP/OCI gateway on {} runs WITHOUT TLS — credentials and \
                 artifacts travel in cleartext. Set ron_tls for any non-loopback use.",
                addr
            );
        }

        let acceptor = settings.tls.clone().map(TlsAcceptor::from);

        loop {
            let (stream, peer) = match listener.accept().await {
                Ok(conn) => conn,
                Err(e) => {
                    log::debug!("gateway accept error: {}", e);
                    continue;
                }
            };
            let peer = peer.to_string();
            let settings = settings.clone();
            let acceptor = acceptor.clone();
            tokio::spawn(async move {
                match acceptor {
                    Some(acceptor) => {
                        let tls_stream = match acceptor.accept(stream).await {
                            Ok(s) => s,
                            Err(e) => {
                                log::debug!("TLS handshake failed: {}", e);
                                return;
                            }
                        };
                        // Pull the mTLS identity (if any) out of the handshake.
                        let client_cn = tls_stream
                            .get_ref()
                            .1
                            .peer_certificates()
                            .and_then(leaf_common_name);
                        serve(TokioIo::new(tls_stream), settings, client_cn, peer).await;
                    }
                    None => serve(TokioIo::new(stream), settings, None, peer).await,
                }
            });
        }
    });
    Ok(())
}

async fn serve<I>(
    io: TokioIo<I>,
    settings: Arc<GatewaySettings>,
    client_cn: Option<String>,
    peer: String,
) where
    I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + 'static,
{
    let svc = service_fn(move |req| {
        let settings = settings.clone();
        let client_cn = client_cn.clone();
        let peer = peer.clone();
        async move { handle(req, settings, client_cn, peer).await }
    });
    if let Err(e) = ConnBuilder::new(TokioExecutor::new())
        .serve_connection(io, svc)
        .await
    {
        log::debug!("gateway connection error: {}", e);
    }
}

async fn handle(
    req: Request<Incoming>,
    settings: Arc<GatewaySettings>,
    client_cn: Option<String>,
    peer: String,
) -> Result<Response<Full<Bytes>>, std::convert::Infallible> {
    // Request-handling clock: observed into the `holger_request_duration_seconds`
    // histogram at each door's exit, right where the request counter is bumped, so
    // latency is measured over the same handled-request population.
    let started = std::time::Instant::now();
    let method = req.method().as_str().to_owned();
    let path = req.uri().path().to_owned();
    let path_and_query = req
        .uri()
        .path_and_query()
        .map(|pq| pq.as_str().to_owned())
        .unwrap_or_else(|| path.clone());

    // Checksum-verify-on-deploy (Artifactory `X-Checksum-*` convention): a client
    // may advertise the digest of the bytes it is uploading. Capture the strong
    // digests NOW — before `req.into_body()` consumes the request — so the write
    // path can recompute and reject a corrupted upload before it reaches a
    // backend. Header lookups are case-insensitive (canonical lowercase names).
    let want_sha256 = req
        .headers()
        .get("x-checksum-sha256")
        .and_then(|v| v.to_str().ok())
        .map(str::to_owned);
    let want_sha512 = req
        .headers()
        .get("x-checksum-sha512")
        .and_then(|v| v.to_str().ok())
        .map(str::to_owned);

    // Unauthenticated, backend-less endpoints. Probes are not artifact
    // operations, so they are not audited.
    if method == "GET" {
        if path == "/healthz" || path == "/readyz" {
            return Ok(text(200, "ok"));
        }
        if path == "/v2" || path == "/v2/" {
            // OCI version probe — clients hit this before auth.
            return Ok(Response::builder()
                .status(200)
                .header("Docker-Distribution-Api-Version", "registry/2.0")
                .body(Full::new(Bytes::from_static(b"{}")))
                .expect("static response"));
        }
        // Read-only cross-repo SEARCH door (parity §H). Reserved under `/-/` so it
        // never collides with a repository name; open read (anonymous), audited as
        // a List. Query axes come from the query string:
        //   /-/search?name=&namespace=&version=&checksum=&path=&repo=&repo=&limit=
        // Returns the same JSON shape the engine produces (artifacts + paths +
        // truncated). This is the read-only HTTP twin of the `SearchService` RPC.
        if path == "/-/search" {
            let query = parse_search_query(req.uri().query().unwrap_or(""));
            let repos = settings.routes.all_repos();
            let props = settings.properties.clone();
            let results = tokio::task::spawn_blocking(move || {
                crate::search::search_repos_with_properties(
                    &repos,
                    &query,
                    props.as_deref().map(|p| p as &dyn crate::search::PropertyLookup),
                )
            })
            .await;
            let hit_count = results
                .as_ref()
                .map(|r| (r.artifacts.len() + r.paths.len()) as u64)
                .unwrap_or(0);
            let (status, body) = match results {
                Ok(r) => match serde_json::to_vec(&r) {
                    Ok(json) => (200u16, json),
                    Err(e) => (500, format!("serialize error: {e}").into_bytes()),
                },
                Err(e) => (500, format!("search task failed: {e}").into_bytes()),
            };
            // Audit the read (path column carries the raw query string).
            let ev = AuditEvent::new(
                "anonymous",
                AuditAction::List,
                "",
                &path_and_query,
                &peer,
                status,
                hit_count,
            );
            if let Err(e) = settings.audit.record(ev) {
                log::warn!("audit record failed: {e}");
            }
            crate::grpc::functional_status(
                "holger-http/search",
                "search_served",
                status == 200,
                &format!("{hit_count} hits"),
            );
            crate::metrics::global().record_request(
                crate::metrics::Verb::Search,
                status,
                body.len() as u64,
            );
            crate::metrics::global().observe_request_duration(started.elapsed().as_secs_f64());
            let ct = if status == 200 { "application/json" } else { "text/plain" };
            return Ok(Response::builder()
                .status(status)
                .header("Content-Type", ct)
                .body(Full::new(Bytes::from(body)))
                .unwrap_or_else(|_| text(500, "Internal error")));
        }
        // Read-only hosted-SBOM door (parity §6/§11). The read twin of the
        // `SbomService.FetchSbom` RPC — attach stays gRPC/CLI-only. Coordinate
        // comes from the query string:
        //   /-/sbom?repo=&namespace=&name=&version=
        // Serves the hosted CycloneDX document byte-identical (200), or 404 when
        // no SBOM is attached to that coordinate. Open read, audited as a Download.
        if path == "/-/sbom" {
            let (status, ct, body, repo, label) =
                serve_sbom_door(settings.sboms.as_ref(), req.uri().query().unwrap_or(""));
            let ev = AuditEvent::new(
                "anonymous",
                AuditAction::Download,
                &repo,
                &label,
                &peer,
                status,
                if status == 200 { body.len() as u64 } else { 0 },
            );
            if let Err(e) = settings.audit.record(ev) {
                log::warn!("audit record failed: {e}");
            }
            crate::grpc::functional_status(
                "holger-http/sbom",
                "sbom_served",
                status == 200,
                &format!("{status} {label}"),
            );
            crate::metrics::global().record_request(
                crate::metrics::Verb::Sbom,
                status,
                if status == 200 { body.len() as u64 } else { 0 },
            );
            crate::metrics::global().observe_request_duration(started.elapsed().as_secs_f64());
            return Ok(Response::builder()
                .status(status)
                .header("Content-Type", ct)
                .body(Full::new(Bytes::from(body)))
                .unwrap_or_else(|_| text(500, "Internal error")));
        }
        // Read-only custom-PROPERTIES door (parity §8). The read twin of the
        // `property` search axis — property WRITES are CLI/in-process for now.
        //   /-/properties?repo=&namespace=&name=&version=
        // Serves the coordinate's `{ key: [values] }` map as JSON (200, `{}` when
        // none), or 400 when the coordinate is incomplete. Open read, audited.
        if path == "/-/properties" {
            let (status, body, repo, label) =
                serve_properties_door(settings.properties.as_ref(), req.uri().query().unwrap_or(""));
            let ev = AuditEvent::new(
                "anonymous",
                AuditAction::List,
                &repo,
                &label,
                &peer,
                status,
                0,
            );
            if let Err(e) = settings.audit.record(ev) {
                log::warn!("audit record failed: {e}");
            }
            crate::grpc::functional_status(
                "holger-http/properties",
                "properties_served",
                status == 200,
                &format!("{status} {label}"),
            );
            let ct = if status == 200 { "application/json" } else { "text/plain" };
            return Ok(Response::builder()
                .status(status)
                .header("Content-Type", ct)
                .body(Full::new(Bytes::from(body)))
                .unwrap_or_else(|_| text(500, "Internal error")));
        }
        // Read-only Prometheus scrape door (parity §K). Reserved (no `/-/` prefix
        // — Prometheus scrapes the conventional `/metrics`), unauthenticated and
        // unaudited (it exposes only aggregate registry counters, not artifacts).
        // Counters come from the process-global registry every verb bumps; the
        // per-repo artifact gauge lists each backend at scrape time, so it runs
        // under `spawn_blocking` off the async reactor.
        if path == "/metrics" {
            let repos = settings.routes.all_repos();
            let body = tokio::task::spawn_blocking(move || {
                let repo_counts: Vec<(String, usize)> = repos
                    .iter()
                    .map(|(name, backend)| {
                        // Bounded listing: the gauge is a health signal, not a
                        // census, so a huge repo caps at METRICS_LIST_CAP and an
                        // errored backend reports 0 rather than failing the scrape.
                        let n = backend
                            .list(None, METRICS_LIST_CAP)
                            .map(|v| v.len())
                            .unwrap_or(0);
                        (name.clone(), n)
                    })
                    .collect();
                crate::metrics::global().render_prometheus(&repo_counts)
            })
            .await;
            let (status, body) = match body {
                Ok(text) => (200u16, text.into_bytes()),
                Err(e) => (500u16, format!("metrics task failed: {e}").into_bytes()),
            };
            crate::grpc::functional_status(
                "holger-http/metrics",
                "metrics_served",
                status == 200,
                &format!("{} bytes", body.len()),
            );
            return Ok(Response::builder()
                .status(status)
                // Prometheus text exposition format 0.0.4.
                .header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
                .body(Full::new(Bytes::from(body)))
                .unwrap_or_else(|_| text(500, "Internal error")));
        }
    }

    // Custom-PROPERTIES WRITE door (parity §8): set/remove a property remotely —
    // the write twin of the read door above + the gRPC `PropertyService.SetProperty`
    // RPC (property write was CLI/in-process only before this).
    //   PUT|POST /-/properties?repo=&namespace=&name=&version=&key=&value=&value=…
    //   DELETE   /-/properties?repo=&…&key=                       (remove the key)
    // Write-gated (writer/admin), SCOPED to the TARGET repo from the query string
    // (not the `-` path segment), mirroring the gRPC write gate. Values ride in
    // repeatable `value=` params, so there is no artifact body / checksum gate.
    if path == "/-/properties" && matches!(method.as_str(), "PUT" | "POST" | "DELETE") {
        let raw_q = req.uri().query().unwrap_or("").to_owned();
        let coord = parse_sbom_coord(&raw_q);
        // AuthN: resolve identity (OIDC sub / mTLS CN), or `None` when auth is open.
        let bearer = req
            .headers()
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "));
        let identity =
            match auth::validate_request(&settings.auth_config, bearer, client_cn.as_deref()).await {
                Ok(id) => id,
                Err(_) => return Ok(text(401, "Unauthorized")),
            };
        // AuthZ (writer/admin), scoped to the coordinate's repository.
        if let Err((status, detail)) =
            authorize_http_write(&settings.auth_config, identity.as_ref(), &method, &coord.repository)
        {
            let who = identity.as_ref().map(|i| i.subject.as_str()).unwrap_or("anonymous");
            let ev = AuditEvent::new(who, AuditAction::Upload, &coord.repository, &path, &peer, status, 0)
                .with_detail(detail.to_string());
            if let Err(e) = settings.audit.record(ev) {
                log::warn!("audit record failed: {e}");
            }
            let msg = if status == 403 { "Forbidden" } else { "Unauthorized" };
            return Ok(text(status, msg));
        }
        let who = identity.map(|i| i.subject).unwrap_or_else(|| "anonymous".to_string());
        let (status, body, repo, label) =
            serve_properties_write(settings.properties.as_ref(), &method, &raw_q);
        let action = if method == "DELETE" { AuditAction::Delete } else { AuditAction::Upload };
        let ev = AuditEvent::new(&who, action, &repo, &label, &peer, status, 0);
        if let Err(e) = settings.audit.record(ev) {
            log::warn!("audit record failed: {e}");
        }
        crate::grpc::functional_status(
            "holger-http/properties_write",
            "property_written",
            status == 200,
            &format!("{status} {label}"),
        );
        crate::metrics::global().record_request(
            if method == "DELETE" { crate::metrics::Verb::Delete } else { crate::metrics::Verb::Put },
            status,
            0,
        );
        crate::metrics::global().observe_request_duration(started.elapsed().as_secs_f64());
        let ct = if status == 200 { "application/json" } else { "text/plain" };
        return Ok(Response::builder()
            .status(status)
            .header("Content-Type", ct)
            .body(Full::new(Bytes::from(body)))
            .unwrap_or_else(|_| text(500, "Internal error")));
    }

    // Map the HTTP verb to an audit action: reads (GET/HEAD) are downloads,
    // DELETE removes, everything else (PUT/POST/PATCH) stores.
    let is_read = method == "GET" || method == "HEAD";
    let action = if is_read {
        AuditAction::Download
    } else if method == "DELETE" {
        AuditAction::Delete
    } else {
        AuditAction::Upload
    };
    // Repository name for the audit record: the first OCI/classic path segment.
    let repo_key = route_key(&path).unwrap_or_default().to_owned();
    // Prometheus verb for this request, via the ONE shared audit-action→verb
    // classifier (Law #5 — same mapping the gRPC `record_audit` choke point uses).
    // The artifact door only ever produces Download/Upload/Delete here, all of
    // which classify; `Get` is a safe default that can't be reached.
    let metric_verb = crate::metrics::Verb::from_audit_action(action).unwrap_or(crate::metrics::Verb::Get);
    // Best-effort audit append, mirroring the gRPC `record_audit` helper. Every
    // outcome also bumps the process-global metrics registry (`/metrics`), so a
    // verb, its bytes served, and any 5xx are counted at the one exit point.
    let record = |ident: &str, status: u16, bytes: u64, detail: &str| {
        crate::metrics::global().record_request(metric_verb, status, bytes);
        crate::metrics::global().observe_request_duration(started.elapsed().as_secs_f64());
        let mut ev =
            AuditEvent::new(ident, action, &repo_key, &path, &peer, status, bytes);
        if !detail.is_empty() {
            ev = ev.with_detail(detail);
        }
        if let Err(e) = settings.audit.record(ev) {
            log::warn!("audit record failed: {e}");
        }
    };

    // Writes require a valid identity (Bearer or mTLS CN) and, when role policy
    // is configured, authorization. Reads stay open and record as the anonymous
    // principal.
    let ident = if is_read {
        "anonymous".to_string()
    } else {
        let bearer = req
            .headers()
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "));
        // AuthN: resolve the identity (OIDC sub / mTLS CN), or `None` when auth is
        // open (no methods configured).
        let identity =
            match auth::validate_request(&settings.auth_config, bearer, client_cn.as_deref()).await
            {
                Ok(id) => id,
                Err(_) => {
                    record("anonymous", 401, 0, "unauthorized");
                    return Ok(text(401, "Unauthorized"));
                }
            };

        // AuthZ (RBAC) — mirrors `grpc::authorize_write`, scoped to the target repo
        // so a per-repo ACL (`repo_roles`) governs this write.
        if let Err((status, detail)) =
            authorize_http_write(&settings.auth_config, identity.as_ref(), &method, &repo_key)
        {
            let who = identity
                .as_ref()
                .map(|i| i.subject.as_str())
                .unwrap_or("anonymous");
            record(who, status, 0, detail);
            let msg = if status == 403 { "Forbidden" } else { "Unauthorized" };
            return Ok(text(status, msg));
        }

        identity
            .map(|i| i.subject)
            .unwrap_or_else(|| "anonymous".to_string())
    };

    let backend = match routes_lookup(&settings.routes, &path) {
        Some(b) => b,
        None => {
            record(&ident, 404, 0, "unknown repository");
            return Ok(text(404, "Unknown repository"));
        }
    };

    // Serve-time quarantine gate (property-policy §): on a READ, if the backend can
    // map this path to a coordinate and that coordinate's properties match the
    // configured quarantine predicate (e.g. quarantine=true), refuse with 404 —
    // served as absent so a bad/unverified blob is blocked without deleting it. Off
    // unless configured, so the default serve path pays nothing.
    if is_read
        && serve_quarantined(
            settings.quarantine.as_ref(),
            settings.properties.as_ref(),
            backend.as_ref(),
            &path,
            &repo_key,
        )
    {
        record(&ident, 404, 0, "quarantined — refused at serve boundary");
        crate::grpc::functional_status("holger-http/quarantine", "quarantine_blocked", true, &path);
        return Ok(text(404, "Not found"));
    }

    // Read-only repository write-gate, mirroring grpc.rs: reject writes to a
    // non-writable backend BEFORE collecting the body (fail closed, no upload
    // DoS surface). The gRPC path returns 403 here; the HTTP path used to defer
    // to the backend, which silently accepted/ignored the write.
    if !is_read && !backend.is_writable() {
        record(&ident, 403, 0, "repository is read-only");
        return Ok(text(403, "Forbidden"));
    }

    // Cap the body to bound memory (upload DoS guard).
    let body = match Limited::new(req.into_body(), settings.max_body_bytes)
        .collect()
        .await
    {
        Ok(collected) => collected.to_bytes(),
        Err(_) => {
            record(&ident, 413, 0, "payload too large");
            return Ok(text(413, "Payload too large"));
        }
    };

    // Integrity gate: if the client advertised a digest, it MUST match the bytes
    // we received or the write is rejected (400) before any backend sees it —
    // holger's "the bytes are what you think they are" guarantee at the door.
    if !is_read {
        if let Err((status, detail)) =
            verify_upload_checksum(want_sha256.as_deref(), want_sha512.as_deref(), &body)
        {
            record(&ident, status, body.len() as u64, detail);
            return Ok(text(status, detail));
        }
    }

    let dispatch = tokio::task::spawn_blocking(move || {
        backend.handle_http2_request(&method, &path_and_query, &body)
    })
    .await;

    match dispatch {
        Ok(Ok((status, headers, body))) => {
            record(&ident, status, body.len() as u64, "");
            let mut builder = Response::builder().status(status);
            for (k, v) in headers {
                builder = builder.header(k, v);
            }
            Ok(builder
                .body(Full::new(Bytes::from(body)))
                .unwrap_or_else(|_| text(500, "Internal error")))
        }
        // Generic message to the client; detail stays in the server log.
        Ok(Err(e)) => {
            record(&ident, 500, 0, &e.to_string());
            log::warn!("backend error for {}: {:#}", path, e);
            Ok(text(500, "Internal error"))
        }
        Err(_) => {
            record(&ident, 500, 0, "dispatch task failed");
            Ok(text(500, "Internal error"))
        }
    }
}

/// Checksum-verify-on-deploy. Recomputes the strong digest(s) the client
/// advertised (`X-Checksum-Sha256` / `X-Checksum-Sha512`, Artifactory's
/// checksum-deploy convention) over the received bytes and returns
/// `Err((400, detail))` on any mismatch. Verification is **opt-in**: a missing
/// or empty header is `Ok(())` (the write proceeds unverified, as before). Only
/// strong digests are honoured — legacy `sha1`/`md5` are intentionally
/// unsupported (too weak to be an integrity guarantee). Hex compare is
/// case-insensitive.
fn verify_upload_checksum(
    want_sha256: Option<&str>,
    want_sha512: Option<&str>,
    body: &[u8],
) -> Result<(), (u16, &'static str)> {
    use sha2::{Digest, Sha256, Sha512};
    if let Some(expected) = want_sha256 {
        let expected = expected.trim();
        if !expected.is_empty() {
            let got = hex::encode(Sha256::digest(body));
            if !got.eq_ignore_ascii_case(expected) {
                return Err((400, "sha256 checksum mismatch"));
            }
        }
    }
    if let Some(expected) = want_sha512 {
        let expected = expected.trim();
        if !expected.is_empty() {
            let got = hex::encode(Sha512::digest(body));
            if !got.eq_ignore_ascii_case(expected) {
                return Err((400, "sha512 checksum mismatch"));
            }
        }
    }
    Ok(())
}

/// Percent-decode an `application/x-www-form-urlencoded` component: `+` → space
/// and `%XX` → the byte. Invalid escapes are passed through verbatim (a search
/// term is best-effort, never a hard error). Kept tiny + dependency-free.
fn form_decode(s: &str) -> String {
    let bytes = s.as_bytes();
    let mut out: Vec<u8> = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        match bytes[i] {
            b'+' => {
                out.push(b' ');
                i += 1;
            }
            b'%' if i + 2 < bytes.len() => {
                let hex = |c: u8| -> Option<u8> {
                    match c {
                        b'0'..=b'9' => Some(c - b'0'),
                        b'a'..=b'f' => Some(c - b'a' + 10),
                        b'A'..=b'F' => Some(c - b'A' + 10),
                        _ => None,
                    }
                };
                match (hex(bytes[i + 1]), hex(bytes[i + 2])) {
                    (Some(h), Some(l)) => {
                        out.push(h << 4 | l);
                        i += 3;
                    }
                    _ => {
                        out.push(b'%');
                        i += 1;
                    }
                }
            }
            b => {
                out.push(b);
                i += 1;
            }
        }
    }
    String::from_utf8_lossy(&out).into_owned()
}

/// Build a [`search::SearchQuery`](crate::search::SearchQuery) from a raw URL
/// query string (`name=serde&version=1.0.0&repo=rust-dev&repo=maven-dev&limit=50`).
/// Recognised keys: `name`, `namespace` (alias `group`), `version`, `checksum`
/// (alias `sha256`), `path`, `repo`/`repository` (repeatable), `limit`. Empty
/// values are treated as unset; unknown keys are ignored.
fn parse_search_query(raw: &str) -> crate::search::SearchQuery {
    let mut q = crate::search::SearchQuery::default();
    for pair in raw.split('&').filter(|s| !s.is_empty()) {
        let (key, val) = match pair.split_once('=') {
            Some((k, v)) => (k, form_decode(v)),
            None => (pair, String::new()),
        };
        let set = |slot: &mut Option<String>, v: String| {
            if !v.is_empty() {
                *slot = Some(v);
            }
        };
        match key {
            "name" => set(&mut q.name, val),
            "namespace" | "group" => set(&mut q.namespace, val),
            "version" => set(&mut q.version, val),
            "checksum" | "sha256" => set(&mut q.checksum, val),
            "path" => set(&mut q.path, val),
            // `property=key:value` (repeatable) — a custom-property filter. The
            // key/value split is on the FIRST ':' so a value may itself contain
            // ':'; `property=key` (no ':') means "key present, any value".
            "property" | "prop" => {
                if !val.is_empty() {
                    let (k, v) = match val.split_once(':') {
                        Some((k, v)) => (k.to_string(), v.to_string()),
                        None => (val, String::new()),
                    };
                    if !k.is_empty() {
                        q.properties.push((k, v));
                    }
                }
            }
            "repo" | "repository" => {
                if !val.is_empty() {
                    q.repositories.push(val);
                }
            }
            "limit" => {
                if let Ok(n) = val.parse::<usize>() {
                    q.limit = n;
                }
            }
            _ => {}
        }
    }
    q
}

/// The parsed coordinate for the `GET /-/sbom` door: which repository +
/// artifact `(namespace/name/version)` an SBOM is requested for. Missing pieces
/// are empty strings (the handler rejects a query lacking `repo`/`name`/`version`).
pub(crate) struct SbomCoordQuery {
    pub repository: String,
    pub id: traits::ArtifactId,
}

/// Build an [`SbomCoordQuery`] from a raw URL query string
/// (`repo=rust-dev&name=serde&version=1.0.0[&namespace=…]`). Keys: `repo`
/// (alias `repository`), `namespace` (alias `group`), `name`, `version`. Empty
/// values are unset; an empty namespace ⇒ `None` (a namespace-less coordinate).
fn parse_sbom_coord(raw: &str) -> SbomCoordQuery {
    let (mut repository, mut namespace, mut name, mut version) =
        (String::new(), String::new(), String::new(), String::new());
    for pair in raw.split('&').filter(|s| !s.is_empty()) {
        let (key, val) = match pair.split_once('=') {
            Some((k, v)) => (k, form_decode(v)),
            None => (pair, String::new()),
        };
        match key {
            "repo" | "repository" => repository = val,
            "namespace" | "group" => namespace = val,
            "name" => name = val,
            "version" => version = val,
            _ => {}
        }
    }
    SbomCoordQuery {
        repository,
        id: traits::ArtifactId {
            namespace: if namespace.is_empty() { None } else { Some(namespace) },
            name,
            version,
        },
    }
}

/// Serve the read-only `GET /-/sbom` door: resolve the coordinate from
/// `raw_query`, fetch the hosted SBOM from `sboms`, and return
/// `(status, content_type, body, repository, label)`. **Pure** over the store it
/// is given, so the HTTP handler and its test share exactly one code path — the
/// byte-identical serving is asserted directly, not through a live socket.
///
/// * 400 when `repo`/`name`/`version` are not all present.
/// * 200 + the byte-identical CycloneDX document when an SBOM is attached.
/// * 404 when no SBOM is attached (or hosting is not configured on this server).
/// * 500 on a store read error.
pub(crate) fn serve_sbom_door(
    sboms: Option<&crate::sbom::SharedSbomStore>,
    raw_query: &str,
) -> (u16, &'static str, Vec<u8>, String, String) {
    let coord = parse_sbom_coord(raw_query);
    let label = format!(
        "{}/{}@{}",
        coord.id.namespace.as_deref().unwrap_or("-"),
        coord.id.name,
        coord.id.version,
    );
    let (status, ct, body): (u16, &'static str, Vec<u8>) = if coord.repository.is_empty()
        || coord.id.name.is_empty()
        || coord.id.version.is_empty()
    {
        (400, "text/plain", b"sbom query needs repo, name and version".to_vec())
    } else {
        match sboms.map(|s| s.fetch(&coord.repository, &coord.id)) {
            Some(Ok(Some(doc))) => (200, "application/json", doc),
            // No SBOM attached, or hosting not configured on this server.
            Some(Ok(None)) | None => {
                (404, "text/plain", b"no SBOM hosted for that coordinate".to_vec())
            }
            Some(Err(e)) => (500, "text/plain", format!("sbom read error: {e}").into_bytes()),
        }
    };
    (status, ct, body, coord.repository, label)
}

/// Serve the read-only `GET /-/properties` door: resolve the coordinate from
/// `raw_query` (reusing the SBOM coordinate parser), read its property map from
/// `props`, and return `(status, json_body, repository, label)`. **Pure** over
/// the store it is given, so the handler and its test share one code path.
///
/// * 400 when `repo`/`name`/`version` are not all present.
/// * 200 + the coordinate's `{ key: [values] }` JSON (`{}` when none, or when the
///   store is not configured — an unset coordinate has no properties either way).
pub(crate) fn serve_properties_door(
    props: Option<&crate::properties::SharedPropertyStore>,
    raw_query: &str,
) -> (u16, Vec<u8>, String, String) {
    let coord = parse_sbom_coord(raw_query);
    let label = format!(
        "{}/{}@{}",
        coord.id.namespace.as_deref().unwrap_or("-"),
        coord.id.name,
        coord.id.version,
    );
    if coord.repository.is_empty() || coord.id.name.is_empty() || coord.id.version.is_empty() {
        return (
            400,
            b"properties query needs repo, name and version".to_vec(),
            coord.repository,
            label,
        );
    }
    let map = props
        .map(|p| p.get(&coord.repository, &coord.id))
        .unwrap_or_default();
    let body = serde_json::to_vec(&map).unwrap_or_else(|_| b"{}".to_vec());
    (200, body, coord.repository, label)
}

/// Serve a `PUT`/`POST`/`DELETE /-/properties` WRITE: parse the coordinate + `key`
/// (+ repeatable `value=` for a set) from `raw_query`, apply it to `props`, and
/// return `(status, json_body, repository, label)`. **Pure** over the store, so
/// the handler and its test share one code path (the auth gate runs in the
/// handler, before this).
///
/// * 400 when `repo`/`name`/`version`/`key` are not all present.
/// * `DELETE`, or `PUT`/`POST` with no `value=`, **removes** the key.
/// * `PUT`/`POST` with values **sets** them (last-wins, deduped).
/// * 200 + the coordinate's resulting `{ key: [values] }` JSON.
/// * 503 when no property store is configured on this server.
pub(crate) fn serve_properties_write(
    props: Option<&crate::properties::SharedPropertyStore>,
    method: &str,
    raw_query: &str,
) -> (u16, Vec<u8>, String, String) {
    let coord = parse_sbom_coord(raw_query);
    // Pull `key` + repeatable `value=` from the same query string.
    let (mut key, mut values) = (String::new(), Vec::<String>::new());
    for pair in raw_query.split('&').filter(|s| !s.is_empty()) {
        let (k, v) = match pair.split_once('=') {
            Some((k, v)) => (k, form_decode(v)),
            None => (pair, String::new()),
        };
        match k {
            "key" => key = v,
            "value" => {
                if !v.is_empty() {
                    values.push(v);
                }
            }
            _ => {}
        }
    }
    let label = format!(
        "{}/{}@{}",
        coord.id.namespace.as_deref().unwrap_or("-"),
        coord.id.name,
        coord.id.version,
    );
    if coord.repository.is_empty()
        || coord.id.name.is_empty()
        || coord.id.version.is_empty()
        || key.is_empty()
    {
        return (
            400,
            b"properties write needs repo, name, version and key".to_vec(),
            coord.repository,
            label,
        );
    }
    let store = match props {
        Some(s) => s,
        None => {
            return (503, b"property store not configured".to_vec(), coord.repository, label)
        }
    };
    // DELETE (or a set with no values) removes the key; otherwise set the values.
    let effective = if method == "DELETE" { Vec::new() } else { values };
    match store.set(&coord.repository, &coord.id, &key, effective) {
        Ok(map) => {
            let body = serde_json::to_vec(&map).unwrap_or_else(|_| b"{}".to_vec());
            (200, body, coord.repository, label)
        }
        Err(e) => (500, format!("property write error: {e}").into_bytes(), coord.repository, label),
    }
}

/// The repository key a request path resolves to. `/v2/{name}/…` keys by the
/// first OCI name segment; everything else by the first path segment. `None`
/// when the path has no segments.
fn route_key(path: &str) -> Option<&str> {
    let segs: Vec<&str> = path.trim_start_matches('/').split('/').filter(|s| !s.is_empty()).collect();
    match segs.as_slice() {
        ["v2", name, ..] => Some(*name),
        [first, ..] => Some(*first),
        [] => None,
    }
}

/// Authorization decision for an HTTP write/delete, mirroring
/// [`grpc::authorize_write`](crate::grpc). `identity` is the resolved principal
/// (`None` = no auth methods configured / open access). RBAC is inactive unless
/// role policy is configured; when active, the resolved role must permit the
/// action — DELETE requires admin (per the role model: admin = promote/delete),
/// other writes (PUT/POST/PATCH) require writer-or-admin — and a configured
/// policy with no authenticated identity fails closed. Returns `Ok(())` to
/// proceed, or `Err((status, detail))` to reject.
fn authorize_http_write(
    auth_config: &AuthConfig,
    identity: Option<&auth::AuthIdentity>,
    method: &str,
    repo: &str,
) -> Result<(), (u16, &'static str)> {
    if !auth_config.rbac_enabled() {
        return Ok(());
    }
    match identity {
        Some(id) => {
            let role = auth_config.role_for_repo(repo, &id.subject);
            let permitted = if method == "DELETE" {
                role.can_admin()
            } else {
                role.can_write()
            };
            crate::grpc::functional_status(
                "holger-http/authorize_http_write",
                "rbac_write_permitted",
                permitted,
                method,
            );
            if permitted {
                Ok(())
            } else {
                Err((403, "insufficient role"))
            }
        }
        None => {
            crate::grpc::functional_status(
                "holger-http/authorize_http_write",
                "rbac_write_permitted",
                false,
                "rbac configured but no authenticated identity",
            );
            Err((401, "authentication required"))
        }
    }
}

/// Resolve the backend for a request path. `/v2/{name}/…` routes by the first
/// OCI name segment; everything else by the first path segment.
fn routes_lookup(
    routes: &FastRoutes,
    path: &str,
) -> Option<std::sync::Arc<dyn traits::RepositoryBackendTrait>> {
    routes.lookup(route_key(path)?).cloned()
}

/// Serve-time quarantine predicate for the HTTP read path: does the artifact this
/// `path` addresses carry the configured quarantine property (so it must be refused
/// with 404)? `false` — never gated — unless BOTH a quarantine predicate and a
/// property store are configured AND `backend` can map `path` back to a coordinate
/// (see [`RepositoryBackendTrait::coordinate_for_path`]). Uses the same
/// [`properties::map_matches`](crate::properties::map_matches) rule as the gRPC
/// [`HolgerGrpc::is_quarantined`](crate::grpc::HolgerGrpc) gate, so `quarantine=true`
/// means the same thing on both doors (Law #5, no twin).
fn serve_quarantined(
    quarantine: Option<&(String, String)>,
    properties: Option<&crate::properties::SharedPropertyStore>,
    backend: &dyn traits::RepositoryBackendTrait,
    path: &str,
    repo_key: &str,
) -> bool {
    match (quarantine, properties) {
        (Some((key, value)), Some(store)) => match backend.coordinate_for_path(path) {
            Some(id) => crate::properties::map_matches(&store.get(repo_key, &id), key, value),
            None => false,
        },
        _ => false,
    }
}

fn text(status: u16, msg: &str) -> Response<Full<Bytes>> {
    Response::builder()
        .status(status)
        .header("Content-Type", "text/plain")
        .body(Full::new(Bytes::from(msg.to_owned())))
        .expect("static response")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::auth::{AuthConfig, AuthIdentity, Role};
    use std::collections::HashMap;

    fn ident(sub: &str) -> AuthIdentity {
        AuthIdentity { subject: sub.to_string(), method: "mtls".into() }
    }

    fn cfg_with_roles(pairs: &[(&str, Role)], default_role: Option<Role>) -> AuthConfig {
        let mut roles = HashMap::new();
        for (s, r) in pairs {
            roles.insert((*s).to_string(), *r);
        }
        AuthConfig { methods: vec![], roles, default_role, ..Default::default() }
    }

    // A repo with no per-repo ACL → the global policy governs (unchanged behaviour).
    const R: &str = "any-repo";

    /// RBAC off (no role policy): the HTTP gateway stays authN-only — any
    /// resolved identity, and even an open-auth `None`, may write. Preserves the
    /// pre-RBAC behaviour, matching `grpc::authorize_write`.
    #[test]
    fn rbac_off_allows_all_writes() {
        let cfg = AuthConfig::default();
        assert!(authorize_http_write(&cfg, Some(&ident("alice")), "PUT", R).is_ok());
        assert!(authorize_http_write(&cfg, None, "PUT", R).is_ok());
        assert!(authorize_http_write(&cfg, None, "DELETE", R).is_ok());
    }

    /// With RBAC on, a `reader` must NOT be able to PUT/POST/DELETE through the
    /// HTTP gateway (the bug: writes were authenticated but never authorized).
    #[test]
    fn rbac_on_reader_cannot_write_or_delete() {
        let cfg = cfg_with_roles(&[("bob", Role::Reader)], None);
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("bob")), "PUT", R),
            Err((403, "insufficient role"))
        );
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("bob")), "POST", R),
            Err((403, "insufficient role"))
        );
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("bob")), "DELETE", R),
            Err((403, "insufficient role"))
        );
    }

    /// A `writer` may upload but NOT delete (delete is an admin action); an
    /// `admin` may do both.
    #[test]
    fn rbac_on_writer_uploads_admin_deletes() {
        let cfg = cfg_with_roles(&[("wendy", Role::Writer), ("alice", Role::Admin)], None);
        assert!(authorize_http_write(&cfg, Some(&ident("wendy")), "PUT", R).is_ok());
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("wendy")), "DELETE", R),
            Err((403, "insufficient role")),
            "writer must not delete"
        );
        assert!(authorize_http_write(&cfg, Some(&ident("alice")), "PUT", R).is_ok());
        assert!(authorize_http_write(&cfg, Some(&ident("alice")), "DELETE", R).is_ok());
    }

    /// RBAC configured but the write arrived with no authenticated identity:
    /// fail closed (401), never fall open.
    #[test]
    fn rbac_on_anonymous_write_fails_closed() {
        let cfg = cfg_with_roles(&[("alice", Role::Admin)], None);
        assert_eq!(
            authorize_http_write(&cfg, None, "PUT", R),
            Err((401, "authentication required"))
        );
        // An unmapped identity falls back to the least-privilege Reader → denied.
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("stranger")), "PUT", R),
            Err((403, "insufficient role"))
        );
    }

    /// `default_role: Writer` lets unmapped identities upload but still not
    /// delete — confirms the default-role path is honoured.
    #[test]
    fn rbac_default_role_writer() {
        let cfg = cfg_with_roles(&[], Some(Role::Writer));
        assert!(cfg.rbac_enabled());
        assert!(authorize_http_write(&cfg, Some(&ident("anyone")), "PUT", R).is_ok());
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("anyone")), "DELETE", R),
            Err((403, "insufficient role"))
        );
    }

    /// Per-repo ACL: a subject the GLOBAL policy makes a Reader is a Writer on the
    /// repo its ACL names — and only there. The HTTP write-gate honours the
    /// repo-scoped role.
    #[test]
    fn rbac_per_repo_acl_scopes_the_write_gate() {
        let mut cfg = cfg_with_roles(&[("carol", Role::Reader)], None);
        cfg.repo_roles.insert(
            "team-a-dev".to_string(),
            HashMap::from([("carol".to_string(), Role::Writer)]),
        );
        // On the ACL'd repo carol may upload…
        assert!(authorize_http_write(&cfg, Some(&ident("carol")), "PUT", "team-a-dev").is_ok());
        // …but not delete (that's admin, and her per-repo role is only Writer)…
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("carol")), "DELETE", "team-a-dev"),
            Err((403, "insufficient role"))
        );
        // …and on any other repo she is back to the global Reader → denied.
        assert_eq!(
            authorize_http_write(&cfg, Some(&ident("carol")), "PUT", "other-repo"),
            Err((403, "insufficient role"))
        );
    }

    // ── checksum-verify-on-deploy (X-Checksum-*) ────────────────────────────
    use sha2::{Digest as _, Sha512};

    // LAW #5 dedup: SHA-256 comes from the shared `nornir-hash` leaf (edda); only
    // the SHA-512 variant (not in the leaf) still hand-rolls sha2 here.
    fn sha256_hex(b: &[u8]) -> String {
        nornir_hash::sha256_hex(b)
    }
    fn sha512_hex(b: &[u8]) -> String {
        hex::encode(Sha512::digest(b))
    }

    /// No checksum header ⇒ verification is a no-op (opt-in): the write proceeds.
    #[test]
    fn checksum_absent_is_ok() {
        assert!(verify_upload_checksum(None, None, b"anything").is_ok());
    }

    /// An empty header value is treated as "not supplied" (opt-in), not a mismatch.
    #[test]
    fn checksum_empty_header_is_ok() {
        assert!(verify_upload_checksum(Some(""), Some("  "), b"payload").is_ok());
    }

    /// The correct sha256 for the body verifies.
    #[test]
    fn checksum_sha256_match_ok() {
        let body = b"the artifact bytes";
        assert!(verify_upload_checksum(Some(&sha256_hex(body)), None, body).is_ok());
    }

    /// Uppercase hex still matches (case-insensitive compare).
    #[test]
    fn checksum_sha256_uppercase_ok() {
        let body = b"MiXeD case digest";
        let up = sha256_hex(body).to_uppercase();
        assert!(verify_upload_checksum(Some(&up), None, body).is_ok());
    }

    /// A wrong sha256 is rejected (400) — the write never reaches a backend.
    #[test]
    fn checksum_sha256_mismatch_rejected() {
        let body = b"the artifact bytes";
        let wrong = sha256_hex(b"different bytes entirely");
        assert_eq!(
            verify_upload_checksum(Some(&wrong), None, body),
            Err((400, "sha256 checksum mismatch"))
        );
    }

    /// A garbage / wrong-length digest never accidentally matches.
    #[test]
    fn checksum_sha256_garbage_rejected() {
        assert_eq!(
            verify_upload_checksum(Some("deadbeef"), None, b"x"),
            Err((400, "sha256 checksum mismatch"))
        );
    }

    /// The correct sha512 verifies; a wrong one is rejected.
    #[test]
    fn checksum_sha512_match_and_mismatch() {
        let body = b"sha512 covered payload";
        assert!(verify_upload_checksum(None, Some(&sha512_hex(body)), body).is_ok());
        let wrong = sha512_hex(b"nope");
        assert_eq!(
            verify_upload_checksum(None, Some(&wrong), body),
            Err((400, "sha512 checksum mismatch"))
        );
    }

    /// Both digests supplied and both correct ⇒ Ok.
    #[test]
    fn checksum_both_digests_match_ok() {
        let body = b"double-checked artifact";
        assert!(verify_upload_checksum(
            Some(&sha256_hex(body)),
            Some(&sha512_hex(body)),
            body
        )
        .is_ok());
    }

    // ── /-/search query-string parsing ──────────────────────────────────────

    /// Every recognised axis parses; empty values map to `None`; repeatable
    /// `repo` accumulates; `limit` parses.
    #[test]
    fn search_query_parses_every_axis() {
        let q = parse_search_query(
            "name=serde&group=org.example&version=1.0.0&sha256=ABC&path=%2Ftmp%2Fx&repo=a&repo=b&limit=25",
        );
        assert_eq!(q.name.as_deref(), Some("serde"));
        assert_eq!(q.namespace.as_deref(), Some("org.example"));
        assert_eq!(q.version.as_deref(), Some("1.0.0"));
        assert_eq!(q.checksum.as_deref(), Some("ABC"));
        assert_eq!(q.path.as_deref(), Some("/tmp/x"), "percent-decode applied");
        assert_eq!(q.repositories, vec!["a".to_string(), "b".to_string()]);
        assert_eq!(q.limit, 25);
    }

    /// An empty query string is an all-empty (match-everything) query, and empty
    /// values / unknown keys never fabricate a criterion.
    #[test]
    fn search_query_empty_and_unknown_keys() {
        assert!(parse_search_query("").is_empty());
        let q = parse_search_query("name=&bogus=x&version=");
        assert!(q.is_empty(), "empty values + unknown keys ⇒ no criteria set");
    }

    /// `+` decodes to a space in a search term.
    #[test]
    fn search_query_plus_is_space() {
        let q = parse_search_query("name=hello+world");
        assert_eq!(q.name.as_deref(), Some("hello world"));
    }

    // ── /-/sbom door ────────────────────────────────────────────────────────

    /// Every coordinate axis parses; an empty namespace ⇒ `None`; unknown keys
    /// are ignored.
    #[test]
    fn sbom_coord_parses_axes() {
        let c = parse_sbom_coord("repo=rust-dev&group=org.example&name=serde&version=1.0.0&x=y");
        assert_eq!(c.repository, "rust-dev");
        assert_eq!(c.id.namespace.as_deref(), Some("org.example"));
        assert_eq!(c.id.name, "serde");
        assert_eq!(c.id.version, "1.0.0");
        // No namespace given ⇒ a namespace-less coordinate.
        let c2 = parse_sbom_coord("repo=r&name=n&version=1");
        assert!(c2.id.namespace.is_none());
    }

    /// The HTTP door serves an attached SBOM byte-identical (200) and answers 404
    /// for an unattached / wrong coordinate — the read twin of the RPC round-trip.
    /// The 404 branch is the RED-when-broken guard (a broken key would 200 with
    /// the wrong document).
    #[test]
    fn sbom_door_serves_byte_identical_and_404s_missing() {
        let tmp = tempfile::tempdir().unwrap();
        let store: crate::sbom::SharedSbomStore = Arc::new(crate::sbom::SbomStore::new(tmp.path()));
        let coord = traits::ArtifactId {
            namespace: None,
            name: "serde".into(),
            version: "1.0.0".into(),
        };
        let doc = crate::sbom::build_cyclonedx(&crate::sbom::SbomInput {
            subject: "rust-dev".into(),
            components: vec![],
            vulns: vec![],
        });
        store.attach("rust-dev", &coord, &doc).unwrap();

        // Hit → 200 application/json, byte-identical body.
        let (status, ct, body, repo, _label) =
            serve_sbom_door(Some(&store), "repo=rust-dev&name=serde&version=1.0.0");
        assert_eq!(status, 200);
        assert_eq!(ct, "application/json");
        assert_eq!(body, doc, "the door serves the exact attached SBOM bytes");
        assert_eq!(repo, "rust-dev");

        // Wrong version → 404, never the wrong document.
        let (miss, _, _, _, _) =
            serve_sbom_door(Some(&store), "repo=rust-dev&name=serde&version=2.0.0");
        assert_eq!(miss, 404);

        // Missing repo/name/version → 400.
        let (bad, _, _, _, _) = serve_sbom_door(Some(&store), "name=serde");
        assert_eq!(bad, 400);

        // Hosting not configured (no store) → 404.
        let (none, _, _, _, _) =
            serve_sbom_door(None, "repo=rust-dev&name=serde&version=1.0.0");
        assert_eq!(none, 404);
    }

    // ── /-/properties door + property search axis (§8) ───────────────────────

    /// `property=key:value` (and bare `property=key`) parse into `(key, value)`
    /// filters; the split is on the first ':' so a value may contain ':'.
    #[test]
    fn search_query_parses_property_filters() {
        let q = parse_search_query("name=serde&property=env:prod&prop=team:core&property=reviewed");
        assert_eq!(q.name.as_deref(), Some("serde"));
        assert_eq!(
            q.properties,
            vec![
                ("env".to_string(), "prod".to_string()),
                ("team".to_string(), "core".to_string()),
                ("reviewed".to_string(), String::new()), // bare key = any value
            ]
        );
        // A value containing ':' survives (split on FIRST ':').
        let q2 = parse_search_query("property=url:https://x.y/z");
        assert_eq!(q2.properties, vec![("url".to_string(), "https://x.y/z".to_string())]);
    }

    /// The `/-/properties` door serves a coordinate's property map as JSON, 400s
    /// an incomplete coordinate, and returns `{}` for an unset one (the empty-map
    /// branch is the RED-when-broken guard — a broken key would leak another
    /// coordinate's props).
    #[test]
    fn properties_door_serves_map_and_validates_coordinate() {
        let tmp = tempfile::tempdir().unwrap();
        let store: crate::properties::SharedPropertyStore =
            Arc::new(crate::properties::PropertyStore::new(tmp.path()));
        let coord = traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() };
        store.set("rust-dev", &coord, "env", vec!["prod".into()]).unwrap();

        // Set coordinate → 200 with a JSON object containing the property.
        let (status, body, repo, _label) =
            serve_properties_door(Some(&store), "repo=rust-dev&name=serde&version=1.0.0");
        assert_eq!(status, 200);
        assert_eq!(repo, "rust-dev");
        let json = String::from_utf8(body).unwrap();
        assert!(json.contains("\"env\"") && json.contains("\"prod\""), "map serialized: {json}");

        // Unset coordinate → 200 with an empty object (never another coord's data).
        let (s2, b2, _, _) =
            serve_properties_door(Some(&store), "repo=rust-dev&name=serde&version=2.0.0");
        assert_eq!(s2, 200);
        assert_eq!(String::from_utf8(b2).unwrap(), "{}", "unset coord ⇒ empty map, no leak");

        // Incomplete coordinate → 400.
        let (bad, _, _, _) = serve_properties_door(Some(&store), "name=serde");
        assert_eq!(bad, 400);

        // No store configured → still a valid empty map (never an error).
        let (none, nb, _, _) =
            serve_properties_door(None, "repo=rust-dev&name=serde&version=1.0.0");
        assert_eq!(none, 200);
        assert_eq!(String::from_utf8(nb).unwrap(), "{}");
    }

    /// The `PUT /-/properties` write door sets values, `DELETE` removes, and both
    /// validate the coordinate + key. The set-then-read-back and the remove are
    /// the RED-when-broken guards (a broken write path would 200 but persist
    /// nothing / the wrong thing).
    #[test]
    fn properties_write_door_sets_and_removes() {
        let tmp = tempfile::tempdir().unwrap();
        let store: crate::properties::SharedPropertyStore =
            Arc::new(crate::properties::PropertyStore::new(tmp.path()));

        // PUT with two values → 200, stored.
        let (s, body, repo, _l) = serve_properties_write(
            Some(&store),
            "PUT",
            "repo=rust-dev&name=serde&version=1.0.0&key=env&value=prod&value=staging",
        );
        assert_eq!(s, 200);
        assert_eq!(repo, "rust-dev");
        assert!(String::from_utf8(body).unwrap().contains("staging"), "response echoes the map");
        // Read back through the store: the values persisted.
        let coord = traits::ArtifactId { namespace: None, name: "serde".into(), version: "1.0.0".into() };
        assert_eq!(
            store.get("rust-dev", &coord).get("env").unwrap(),
            &vec!["prod".to_string(), "staging".to_string()]
        );

        // DELETE removes the key.
        let (sd, _b, _r, _l) =
            serve_properties_write(Some(&store), "DELETE", "repo=rust-dev&name=serde&version=1.0.0&key=env");
        assert_eq!(sd, 200);
        assert!(store.get("rust-dev", &coord).is_empty(), "DELETE removed the key");

        // Missing key → 400.
        let (bad, _b, _r, _l) =
            serve_properties_write(Some(&store), "PUT", "repo=rust-dev&name=serde&version=1.0.0");
        assert_eq!(bad, 400);

        // No store configured → 503, never a silent success.
        let (unavail, _b, _r, _l) =
            serve_properties_write(None, "PUT", "repo=r&name=n&version=1&key=k&value=v");
        assert_eq!(unavail, 503);
    }
}