lakehouse/crates/catalogd/src/audit_endpoint.rs

//! `/audit/subject/{id}` HTTP endpoint — legal-tier subject audit response.
//!
//! Specification: docs/specs/SUBJECT_MANIFESTS_ON_CATALOGD.md §4 + §6.
//!
//! Responds to: `GET /subject/{candidate_id}?from=ISO&to=ISO`
//! Auth: requires `X-Lakehouse-Legal-Token` header matching the value in
//! the legal-token file (see LegalAuditAuth). Constant-time compare.
//!
//! Response shape (subject_audit_response.v1):
//!   - manifest: full SubjectManifest from catalogd
//!   - audit_log.rows: SubjectAuditRow entries within [from, to]
//!   - audit_log.chain_verified: bool (HMAC chain verifies end-to-end)
//!   - completeness_attestation: text the lawyer can quote
//!
//! V1 deliberately does NOT include:
//!   - Ed25519 signature on the response — chain verification IS the
//!     v1 attestation; signing is future hardening (spec §10 v2 reserve)
//!   - Live row projection from the underlying datasets — caller can
//!     fetch via /query/sql against the safe_views named in the manifest;
//!     keeping projection out of this endpoint avoids depending on queryd
//!   - External anchor / S3 Object Lock — local HMAC chain is the v1
//!     integrity surface
//!
//! These are documented as future hardening, not silent gaps.

use crate::registry::Registry;
use crate::subject_audit::SubjectAuditWriter;
use axum::{
    extract::{Path, Query, State},
    http::{HeaderMap, StatusCode},
    response::IntoResponse,
    routing::get,
    Json, Router,
};
use serde::{Deserialize, Serialize};
use shared::types::{SubjectAuditRow, SubjectManifest};
use std::sync::Arc;

const LEGAL_TOKEN_HEADER: &str = "x-lakehouse-legal-token";
const RESPONSE_SCHEMA: &str = "subject_audit_response.v1";

/// State for the audit endpoint router. Held outside Registry because
/// the audit writer + legal token aren't part of catalogd's data model
/// — they're sidecar concerns specific to this endpoint.
#[derive(Clone)]
pub struct AuditEndpointState {
    pub registry: Registry,
    pub writer: Arc<SubjectAuditWriter>,
    /// Legal-tier token. Constant-time compared against the header on
    /// every request. None = endpoint returns 503 (audit unavailable).
    pub legal_token: Option<Arc<String>>,
}

impl AuditEndpointState {
    /// Construct from existing pieces. `legal_token_path` is read at
    /// construction and trimmed of trailing whitespace. Missing/empty
    /// = endpoint will 503 every request with a clear reason.
    pub async fn new(
        registry: Registry,
        writer: Arc<SubjectAuditWriter>,
        legal_token_path: &std::path::Path,
    ) -> Self {
        let legal_token = match tokio::fs::read_to_string(legal_token_path).await {
            Ok(s) => {
                let trimmed = s.trim().to_string();
                if trimmed.is_empty() {
                    tracing::warn!(
                        "audit endpoint: legal token file at {} is empty — endpoint will 503",
                        legal_token_path.display()
                    );
                    None
                } else if trimmed.len() < 32 {
                    // 2026-05-03 kimi scrum BLOCK at audit_endpoint.rs:59-60:
                    // 16-byte minimum was too permissive for a legal-tier
                    // secret. Aligned to the SubjectAuditWriter signing-key
                    // minimum (32 bytes — HMAC-SHA256 best practice).
                    tracing::warn!(
                        "audit endpoint: legal token at {} is {} chars (minimum 32) — endpoint will 503",
                        legal_token_path.display(), trimmed.len()
                    );
                    None
                } else {
                    tracing::info!("audit endpoint: legal token loaded from {}", legal_token_path.display());
                    Some(Arc::new(trimmed))
                }
            }
            Err(e) => {
                tracing::warn!(
                    "audit endpoint: legal token unreadable from {} ({e}) — endpoint will 503",
                    legal_token_path.display()
                );
                None
            }
        };
        Self { registry, writer, legal_token }
    }
}

pub fn router(state: AuditEndpointState) -> Router {
    Router::new()
        .route("/subject/{candidate_id}", get(audit_subject))
        .route("/health", get(audit_health))
        .with_state(state)
}

async fn audit_health(State(state): State<AuditEndpointState>) -> impl IntoResponse {
    if state.legal_token.is_some() {
        (StatusCode::OK, "audit endpoint ready").into_response()
    } else {
        (StatusCode::SERVICE_UNAVAILABLE, "audit endpoint disabled (legal token missing)").into_response()
    }
}

#[derive(Deserialize)]
struct AuditQuery {
    /// ISO-8601 start of audit window. None = unbounded back.
    from: Option<chrono::DateTime<chrono::Utc>>,
    /// ISO-8601 end of audit window. None = unbounded forward.
    to: Option<chrono::DateTime<chrono::Utc>>,
}

#[derive(Serialize)]
struct AuditWindowEcho {
    from: Option<chrono::DateTime<chrono::Utc>>,
    to: Option<chrono::DateTime<chrono::Utc>>,
}

#[derive(Serialize)]
struct AuditLogSection {
    rows_in_window: usize,
    rows: Vec<SubjectAuditRow>,
    chain_root: String,
    chain_verified: bool,
    chain_rows_total: usize,
    chain_verification_error: Option<String>,
}

#[derive(Serialize)]
struct AuditResponse {
    schema: &'static str,
    candidate_id: String,
    generated_at: chrono::DateTime<chrono::Utc>,
    generated_by: &'static str,
    request_window: AuditWindowEcho,
    manifest: SubjectManifest,
    audit_log: AuditLogSection,
    /// Datasets the manifest says contain rows for this subject — caller
    /// can issue follow-up queries against the named safe_views to get
    /// the actual row projection. Keeping projection out of this endpoint
    /// avoids depending on queryd from catalogd.
    datasets_referenced: Vec<DatasetRefEcho>,
    safe_views_available: Vec<String>,
    completeness_attestation: String,
}

#[derive(Serialize)]
struct DatasetRefEcho {
    name: String,
    key_column: String,
    key_value: String,
}

/// Constant-time string comparison to avoid timing leaks on the legal
/// token check. `subtle` would be the standard but introducing a dep
/// for one comparison is overkill — manual byte-eq with an unconditional
/// fold over both byte slices.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff: u8 = 0;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Extract + verify the legal-tier token from the request. Returns
/// Err((status, body)) on any failure (missing header, missing token
/// configured, mismatch). Does NOT log token values.
fn require_legal_auth(
    state: &AuditEndpointState,
    headers: &HeaderMap,
) -> Result<(), (StatusCode, &'static str)> {
    let configured = state.legal_token.as_ref().ok_or((
        StatusCode::SERVICE_UNAVAILABLE,
        "audit endpoint disabled — no legal token configured on this gateway",
    ))?;
    let provided = headers.get(LEGAL_TOKEN_HEADER).ok_or((
        StatusCode::UNAUTHORIZED,
        "missing X-Lakehouse-Legal-Token header",
    ))?;
    let provided = provided.to_str().map_err(|_| (
        StatusCode::UNAUTHORIZED,
        "X-Lakehouse-Legal-Token contains non-ASCII characters",
    ))?;
    if !constant_time_eq(provided.as_bytes(), configured.as_bytes()) {
        return Err((StatusCode::UNAUTHORIZED, "X-Lakehouse-Legal-Token mismatch"));
    }
    Ok(())
}

async fn audit_subject(
    State(state): State<AuditEndpointState>,
    Path(candidate_id): Path<String>,
    Query(q): Query<AuditQuery>,
    headers: HeaderMap,
) -> impl IntoResponse {
    if let Err((status, msg)) = require_legal_auth(&state, &headers) {
        return (status, msg).into_response();
    }
    if candidate_id.is_empty() {
        return (StatusCode::BAD_REQUEST, "candidate_id is empty").into_response();
    }

    // 1. Load the subject manifest.
    let manifest = match state.registry.get_subject(&candidate_id).await {
        Some(m) => m,
        None => return (
            StatusCode::NOT_FOUND,
            format!("subject manifest not found for candidate_id={candidate_id}"),
        ).into_response(),
    };

    // 2. Read audit rows in window.
    let rows = match state.writer.read_rows_in_range(&candidate_id, q.from, q.to).await {
        Ok(r) => r,
        Err(e) => return (
            StatusCode::INTERNAL_SERVER_ERROR,
            format!("audit log read failed: {e}"),
        ).into_response(),
    };

    // 3. Verify chain end-to-end (full log, not just the windowed slice;
    //    chain integrity is global, not per-window). chain_root MUST
    //    come from the actual log tip — taking it from the windowed
    //    `rows` would lie when `to` is set or the window catches no
    //    rows. (2026-05-03 opus scrum WARN audit_endpoint.rs:206.)
    let (chain_verified, chain_rows_total, chain_root, chain_err) =
        match state.writer.verify_chain(&candidate_id).await {
            Ok(n) => {
                let tip = state.writer.chain_tip(&candidate_id).await
                    .unwrap_or_else(|| "GENESIS".into());
                (true, n, tip, None)
            }
            Err(e) => (false, 0, String::new(), Some(e)),
        };

    // 4. Tampering detection: if the manifest claims a non-trivial
    //    chain root but the log is empty (chain_rows_total == 0), the
    //    log was deleted between writes. Flag it. (2026-05-03 opus
    //    scrum WARN subject_audit.rs:347 + audit_endpoint coverage.)
    let manifest_root = manifest.audit_log_chain_root.clone();
    let log_was_tampered = chain_verified
        && chain_rows_total == 0
        && !manifest_root.is_empty()
        && manifest_root != "GENESIS";
    let (chain_verified, chain_err) = if log_was_tampered {
        (false, Some(format!(
            "manifest expects chain root '{}' but the audit log has 0 rows; log was deleted or wiped",
            manifest_root,
        )))
    } else {
        (chain_verified, chain_err)
    };

    let datasets_referenced = manifest.datasets.iter()
        .map(|d| DatasetRefEcho {
            name: d.name.clone(),
            key_column: d.key_column.clone(),
            key_value: d.key_value.clone(),
        })
        .collect();
    let safe_views_available = manifest.safe_views.clone();

    let attestation = format!(
        "All audit rows for candidate_id={} within [{}, {}] are included. \
         HMAC chain verified end-to-end: {} rows total, integrity={}.",
        candidate_id,
        q.from.map(|d| d.to_rfc3339()).unwrap_or_else(|| "unbounded".into()),
        q.to.map(|d| d.to_rfc3339()).unwrap_or_else(|| "unbounded".into()),
        chain_rows_total,
        if chain_verified { "verified" } else { "BROKEN" },
    );

    let resp = AuditResponse {
        schema: RESPONSE_SCHEMA,
        candidate_id: candidate_id.clone(),
        generated_at: chrono::Utc::now(),
        generated_by: "catalogd",
        request_window: AuditWindowEcho { from: q.from, to: q.to },
        manifest,
        audit_log: AuditLogSection {
            rows_in_window: rows.len(),
            rows,
            chain_root,
            chain_verified,
            chain_rows_total,
            chain_verification_error: chain_err,
        },
        datasets_referenced,
        safe_views_available,
        completeness_attestation: attestation,
    };

    // Log the legal-tier read so it's discoverable in operations review.
    // Token value is NEVER logged. Only the fact of the read.
    tracing::info!(
        candidate_id = %candidate_id,
        rows_in_window = resp.audit_log.rows_in_window,
        chain_verified = resp.audit_log.chain_verified,
        "legal-tier audit response served"
    );

    (StatusCode::OK, Json(resp)).into_response()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::subject_audit::SubjectAuditWriter;
    use object_store::memory::InMemory;
    use shared::types::{
        AuditAccessor, BiometricConsent, BiometricConsentStatus, ConsentStatus,
        GeneralPiiConsent, SubjectAuditRow, SubjectConsent, SubjectRetention,
        SubjectStatus, SubjectVertical,
    };

    fn fixture_manifest(candidate_id: &str) -> SubjectManifest {
        SubjectManifest {
            schema: "subject_manifest.v1".into(),
            candidate_id: candidate_id.into(),
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
            status: SubjectStatus::Active,
            vertical: SubjectVertical::Unknown,
            consent: SubjectConsent {
                general_pii: GeneralPiiConsent {
                    status: ConsentStatus::PendingBackfillReview,
                    version: String::new(),
                    given_at: None,
                    withdrawn_at: None,
                },
                biometric: BiometricConsent {
                    status: BiometricConsentStatus::NeverCollected,
                    retention_until: None,
                },
            },
            retention: SubjectRetention {
                general_pii_until: chrono::Utc::now() + chrono::Duration::days(365 * 4),
                policy: "test".into(),
            },
            datasets: vec![],
            safe_views: vec![],
            audit_log_path: String::new(),
            audit_log_chain_root: String::new(),
                biometric_collection: None,
        }
    }

    fn fixture_audit_row(candidate_id: &str) -> SubjectAuditRow {
        SubjectAuditRow {
            schema: "subject_audit.v1".into(),
            ts: chrono::Utc::now(),
            candidate_id: candidate_id.into(),
            accessor: AuditAccessor {
                kind: "test".into(),
                daemon: "test".into(),
                purpose: "test".into(),
                trace_id: String::new(),
            },
            fields_accessed: vec!["name".into()],
            result: "success".into(),
            prev_chain_hash: String::new(),
            row_hmac: String::new(),
        }
    }

    async fn fixture_state_with_token(token: &str) -> AuditEndpointState {
        let store: Arc<dyn object_store::ObjectStore> = Arc::new(InMemory::new());
        let registry = Registry::new(store.clone());
        let writer = Arc::new(SubjectAuditWriter::with_inline_key(
            store, (0u8..32).collect(),
        ));
        AuditEndpointState {
            registry,
            writer,
            legal_token: Some(Arc::new(token.into())),
        }
    }

    #[test]
    fn constant_time_eq_basic() {
        assert!(constant_time_eq(b"abc", b"abc"));
        assert!(!constant_time_eq(b"abc", b"abd"));
        assert!(!constant_time_eq(b"abc", b"abcd"));
        assert!(constant_time_eq(b"", b""));
    }

    #[tokio::test]
    async fn missing_legal_token_returns_503() {
        let state = AuditEndpointState {
            registry: Registry::new(Arc::new(InMemory::new())),
            writer: Arc::new(SubjectAuditWriter::with_inline_key(
                Arc::new(InMemory::new()), (0u8..32).collect(),
            )),
            legal_token: None,
        };
        let headers = HeaderMap::new();
        let r = require_legal_auth(&state, &headers);
        assert!(matches!(r, Err((StatusCode::SERVICE_UNAVAILABLE, _))));
    }

    #[tokio::test]
    async fn missing_header_returns_401() {
        let state = fixture_state_with_token("test-token-must-be-long-enough").await;
        let headers = HeaderMap::new();
        let r = require_legal_auth(&state, &headers);
        assert!(matches!(r, Err((StatusCode::UNAUTHORIZED, _))));
    }

    #[tokio::test]
    async fn wrong_token_returns_401() {
        let state = fixture_state_with_token("correct-token-must-be-long-enough").await;
        let mut headers = HeaderMap::new();
        headers.insert(LEGAL_TOKEN_HEADER, "wrong".parse().unwrap());
        let r = require_legal_auth(&state, &headers);
        assert!(matches!(r, Err((StatusCode::UNAUTHORIZED, _))));
    }

    #[tokio::test]
    async fn correct_token_passes_auth() {
        let state = fixture_state_with_token("correct-token-must-be-long-enough").await;
        let mut headers = HeaderMap::new();
        headers.insert(LEGAL_TOKEN_HEADER, "correct-token-must-be-long-enough".parse().unwrap());
        let r = require_legal_auth(&state, &headers);
        assert!(r.is_ok());
    }

    #[tokio::test]
    async fn audit_response_assembly_full_path() {
        let state = fixture_state_with_token("test-token-must-be-long-enough").await;
        // Plant a manifest + 3 audit rows.
        state.registry.put_subject(fixture_manifest("CAND-AUDIT-1")).await.unwrap();
        for _ in 0..3 {
            state.writer.append(fixture_audit_row("CAND-AUDIT-1")).await.unwrap();
        }
        // Read rows in unbounded window.
        let rows = state.writer
            .read_rows_in_range("CAND-AUDIT-1", None, None)
            .await
            .unwrap();
        assert_eq!(rows.len(), 3);
        // Verify chain integrity.
        let n = state.writer.verify_chain("CAND-AUDIT-1").await.unwrap();
        assert_eq!(n, 3);
    }

    #[tokio::test]
    async fn audit_response_window_filters_rows() {
        let state = fixture_state_with_token("test-token-must-be-long-enough").await;
        state.registry.put_subject(fixture_manifest("CAND-WIN-1")).await.unwrap();
        // Append 5 rows.
        let mut row_times = Vec::new();
        for _ in 0..5 {
            let row = fixture_audit_row("CAND-WIN-1");
            row_times.push(row.ts);
            state.writer.append(row).await.unwrap();
            tokio::time::sleep(std::time::Duration::from_millis(2)).await;
        }
        // Window covering only the middle 3 rows.
        let from = row_times[1];
        let to = row_times[3];
        let rows = state.writer
            .read_rows_in_range("CAND-WIN-1", Some(from), Some(to))
            .await
            .unwrap();
        assert!(rows.len() >= 3, "expected ≥3 rows in window; got {}", rows.len());
        assert!(rows.len() <= 3, "expected ≤3 rows in window; got {}", rows.len());
    }

    #[tokio::test]
    async fn empty_token_file_results_in_disabled_endpoint() {
        let tmp = std::env::temp_dir().join(format!("legal_empty_{}.tok", std::process::id()));
        tokio::fs::write(&tmp, "").await.unwrap();
        let store: Arc<dyn object_store::ObjectStore> = Arc::new(InMemory::new());
        let registry = Registry::new(store.clone());
        let writer = Arc::new(SubjectAuditWriter::with_inline_key(store, (0u8..32).collect()));
        let state = AuditEndpointState::new(registry, writer, &tmp).await;
        assert!(state.legal_token.is_none());
        let _ = tokio::fs::remove_file(&tmp).await;
    }

    #[tokio::test]
    async fn short_token_file_rejected_at_load() {
        let tmp = std::env::temp_dir().join(format!("legal_short_{}.tok", std::process::id()));
        tokio::fs::write(&tmp, "tooshort").await.unwrap();
        let store: Arc<dyn object_store::ObjectStore> = Arc::new(InMemory::new());
        let registry = Registry::new(store.clone());
        let writer = Arc::new(SubjectAuditWriter::with_inline_key(store, (0u8..32).collect()));
        let state = AuditEndpointState::new(registry, writer, &tmp).await;
        assert!(state.legal_token.is_none());
        let _ = tokio::fs::remove_file(&tmp).await;
    }
}