athena_rs 3.9.0

use actix_web::http::StatusCode;
use actix_web::{HttpRequest, HttpResponse};
use chrono::Utc;
use sha256::digest;
use sqlx::postgres::PgPool;
use std::collections::HashMap;
use std::net::IpAddr;
use std::sync::Arc;
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::Semaphore;
use tracing::{error, warn};
use uuid::Uuid;

use crate::AppState;
use crate::api::auth::{StaticAdminKeyStatus, extract_gateway_api_key, static_admin_key_status};
use crate::api::response::{forbidden, service_unavailable, unauthorized};
use crate::data::api_key_ips::{
    ApiKeyIpPolicy, IpRule, increment_virgin_request_count, ip_matches_any, load_api_key_ip_policy,
    load_global_ip_rules, mark_virgin_resolved, promote_seen_ips_to_whitelist, record_seen_ip,
};
use crate::data::api_keys::ApiKeySecretRecord;
use crate::data::api_keys::{
    AuthAttemptLogEntry, get_api_key_by_public_id, get_effective_api_key_enforcement,
    insert_auth_attempt_log, touch_api_key_last_used,
};
use crate::parser::query_builder::sanitize_identifier;
use crate::utils::logging_task_limiter::spawn_best_effort_logging_task;
use crate::utils::request_logging::RequestAuthLogContext;

static ENFORCEMENT_CACHE: OnceLock<Mutex<HashMap<String, (bool, Instant)>>> = OnceLock::new();
const ENFORCEMENT_CACHE_TTL: Duration = Duration::from_secs(2);

static IP_POLICY_CACHE: OnceLock<Mutex<HashMap<i64, (ApiKeyIpPolicy, Instant)>>> = OnceLock::new();
static GLOBAL_IP_CACHE: OnceLock<Mutex<HashMap<String, (Vec<IpRule>, Vec<IpRule>, Instant)>>> =
    OnceLock::new();
const IP_POLICY_CACHE_TTL: Duration = Duration::from_secs(2);

#[derive(Debug)]
pub struct GatewayAuthOutcome {
    pub request_id: String,
    pub log_context: RequestAuthLogContext,
    pub bound_client_name: Option<String>,
    pub response: Option<HttpResponse>,
    pub force_deferred_queue: bool,
    pub force_deferred_reason: Option<String>,
}

#[derive(Debug, Clone)]
struct PresentedApiKey {
    public_id: Option<String>,
    secret: Option<String>,
    fingerprint_hash: String,
    fingerprint_salt: String,
}

pub fn read_right_for_resource(resource: Option<&str>) -> String {
    resource_right(resource, "read", "gateway.read")
}

pub fn write_right_for_resource(resource: Option<&str>) -> String {
    resource_right(resource, "write", "gateway.write")
}

pub fn delete_right_for_resource(resource: Option<&str>) -> String {
    resource_right(resource, "delete", "gateway.delete")
}

pub fn query_right() -> String {
    "gateway.query".to_string()
}

/// Right required for `/storage/*` proxy operations (S3-compatible buckets via user-supplied creds).
pub fn storage_proxy_right() -> String {
    "gateway.storage_proxy".to_string()
}

/// Right required for `/typesense/*` proxy/search operations.
pub fn typesense_proxy_right() -> String {
    "gateway.typesense_proxy".to_string()
}

pub fn rpc_right() -> String {
    "gateway.rpc.execute".to_string()
}

fn resource_right(resource: Option<&str>, action: &str, fallback: &str) -> String {
    resource
        .and_then(|value| {
            if value.contains('.') {
                return None;
            }
            sanitize_identifier(value).map(|sanitized| sanitized.trim_matches('"').to_string())
        })
        .map(|value| format!("{}.{}", value, action))
        .unwrap_or_else(|| fallback.to_string())
}

fn split_right(right: &str) -> Option<(&str, &str)> {
    right.split_once('.')
}

fn right_matches(granted: &str, required: &str) -> bool {
    if granted == "*" || granted == required {
        return true;
    }

    let Some((granted_resource, granted_action)) = split_right(granted) else {
        return false;
    };
    let Some((required_resource, required_action)) = split_right(required) else {
        return false;
    };

    if granted_resource == "*" && granted_action == required_action {
        return true;
    }
    if granted_resource == required_resource && granted_action == "*" {
        return true;
    }
    if granted_resource == "gateway" && granted_action == required_action {
        return true;
    }
    if granted_resource == "gateway" && granted_action == "*" {
        return true;
    }

    false
}

fn missing_required_rights(granted_rights: &[String], required_rights: &[String]) -> Vec<String> {
    required_rights
        .iter()
        .filter(|required| {
            !granted_rights
                .iter()
                .any(|granted| right_matches(granted, required))
        })
        .cloned()
        .collect()
}

fn parse_presented_api_key(raw_value: &str) -> PresentedApiKey {
    let fingerprint_salt: String =
        format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
    let fingerprint_hash: String = digest(format!("{}:{}", fingerprint_salt, raw_value));

    let (public_id, secret) = raw_value
        .strip_prefix("ath_")
        .and_then(|value| value.split_once('.'))
        .map(|(public_id, secret)| (Some(public_id.to_string()), Some(secret.to_string())))
        .unwrap_or((None, None));

    PresentedApiKey {
        public_id,
        secret,
        fingerprint_hash,
        fingerprint_salt,
    }
}

fn current_epoch_seconds() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_else(|_| Duration::from_secs(0))
        .as_secs() as i64
}

fn user_agent(req: &HttpRequest) -> Option<String> {
    req.headers()
        .get("User-Agent")
        .and_then(|value| value.to_str().ok())
        .map(str::to_string)
}

fn auth_store_pool(state: &AppState) -> Option<PgPool> {
    state
        .gateway_auth_client_name
        .as_ref()
        .and_then(|name| state.pg_registry.get_pool(name))
}

async fn get_effective_api_key_enforcement_cached(
    pool: &PgPool,
    client_name: Option<&str>,
) -> Result<bool, sqlx::Error> {
    let key: String = format!("{}", client_name.unwrap_or_default());
    let cache = ENFORCEMENT_CACHE.get_or_init(|| Mutex::new(HashMap::new()));

    if let Ok(guard) = cache.lock()
        && let Some((value, inserted_at)) = guard.get(&key)
        && inserted_at.elapsed() < ENFORCEMENT_CACHE_TTL
    {
        return Ok(*value);
    }

    match get_effective_api_key_enforcement(pool, client_name).await {
        Ok(value) => {
            if let Ok(mut guard) = cache.lock() {
                guard.insert(key, (value, Instant::now()));
            }
            Ok(value)
        }
        Err(err) => {
            // Fail-soft: if we have any cached value (even stale), use it to avoid
            // turning transient pool pressure into request-path auth failures.
            if let Ok(guard) = cache.lock()
                && let Some((stale_value, _)) = guard.get(&key)
            {
                warn!(
                    error = %err,
                    client = client_name.unwrap_or_default(),
                    "using stale effective api key enforcement policy due to lookup failure"
                );
                return Ok(*stale_value);
            }
            Err(err)
        }
    }
}

/// Returns the resolved client IP for API-key IP enforcement.
///
/// Honors `X-Real-IP` and (when trusted) the first hop of `X-Forwarded-For`
/// before falling back to the direct peer address.
/// Extracts the apparent client IP from a request.
///
/// Precedence:
/// 1. `X-Real-IP` header (always trusted when present).
/// 2. First entry in `X-Forwarded-For` when `trust_x_forwarded_for` is `true`.
/// 3. `req.peer_addr()` as a fallback.
pub fn extract_client_ip(req: &HttpRequest, trust_x_forwarded_for: bool) -> Option<IpAddr> {
    if let Some(value) = req
        .headers()
        .get("X-Real-IP")
        .and_then(|value| value.to_str().ok())
        && let Ok(ip) = value.trim().parse::<IpAddr>()
    {
        return Some(ip);
    }

    if trust_x_forwarded_for
        && let Some(value) = req
            .headers()
            .get("x-forwarded-for")
            .and_then(|value| value.to_str().ok())
        && let Some(first) = value.split(',').next()
        && let Ok(ip) = first.trim().parse::<IpAddr>()
    {
        return Some(ip);
    }

    req.peer_addr().map(|addr| addr.ip())
}

/// Invalidates the cached IP policy for a key so the next auth reloads it.
pub fn invalidate_ip_policy(api_key_id: i64) {
    if let Some(cache) = IP_POLICY_CACHE.get()
        && let Ok(mut guard) = cache.lock()
    {
        guard.remove(&api_key_id);
    }
}

/// Invalidates the cached global IP rule set for every client scope.
pub fn invalidate_global_ip_rules() {
    if let Some(cache) = GLOBAL_IP_CACHE.get()
        && let Ok(mut guard) = cache.lock()
    {
        guard.clear();
    }
}

async fn get_ip_policy_cached(
    pool: &PgPool,
    api_key_id: i64,
) -> Result<Option<ApiKeyIpPolicy>, sqlx::Error> {
    let cache = IP_POLICY_CACHE.get_or_init(|| Mutex::new(HashMap::new()));

    if let Ok(guard) = cache.lock()
        && let Some((value, inserted_at)) = guard.get(&api_key_id)
        && inserted_at.elapsed() < IP_POLICY_CACHE_TTL
    {
        return Ok(Some(value.clone()));
    }

    match load_api_key_ip_policy(pool, api_key_id).await {
        Ok(Some(policy)) => {
            if let Ok(mut guard) = cache.lock() {
                guard.insert(api_key_id, (policy.clone(), Instant::now()));
            }
            Ok(Some(policy))
        }
        Ok(None) => Ok(None),
        Err(err) => Err(err),
    }
}

async fn get_global_ip_rules_cached(
    pool: &PgPool,
    client_name: Option<&str>,
) -> Result<(Vec<IpRule>, Vec<IpRule>), sqlx::Error> {
    let key: String = client_name.unwrap_or("").to_string();
    let cache = GLOBAL_IP_CACHE.get_or_init(|| Mutex::new(HashMap::new()));

    if let Ok(guard) = cache.lock()
        && let Some((whitelist, blacklist, inserted_at)) = guard.get(&key)
        && inserted_at.elapsed() < IP_POLICY_CACHE_TTL
    {
        return Ok((whitelist.clone(), blacklist.clone()));
    }

    match load_global_ip_rules(pool, client_name).await {
        Ok((whitelist, blacklist)) => {
            if let Ok(mut guard) = cache.lock() {
                guard.insert(key, (whitelist.clone(), blacklist.clone(), Instant::now()));
            }
            Ok((whitelist, blacklist))
        }
        Err(err) => Err(err),
    }
}

/// Outcome of the per-request IP policy check.
#[derive(Debug)]
enum IpPolicyDecision {
    /// Request may proceed.
    Allow,
    /// Request must be rejected; `(reason_code, message, detail)`.
    Deny {
        reason: &'static str,
        message: &'static str,
        detail: String,
    },
}

/// Runs the full IP policy pipeline for an authenticated key.
///
/// Precedence (top wins): global blacklist > per-key blacklist >
/// virgin-mode learning (unresolved keys only) > global whitelist >
/// per-key whitelist > unrestricted.
async fn enforce_ip_policy(
    pool: &PgPool,
    record: &ApiKeySecretRecord,
    client_name: Option<&str>,
    ip: Option<IpAddr>,
) -> IpPolicyDecision {
    let policy: ApiKeyIpPolicy = match get_ip_policy_cached(pool, record.internal_id).await {
        Ok(Some(value)) => value,
        Ok(None) => {
            return IpPolicyDecision::Allow;
        }
        Err(err) => {
            error!(error = %err, "failed to load api key ip policy; allowing request");
            return IpPolicyDecision::Allow;
        }
    };

    let (global_whitelist, global_blacklist) =
        match get_global_ip_rules_cached(pool, client_name).await {
            Ok(value) => value,
            Err(err) => {
                error!(error = %err, "failed to load global ip rules; allowing request");
                (Vec::new(), Vec::new())
            }
        };

    let any_rules_present: bool = !policy.whitelist.is_empty()
        || !policy.blacklist.is_empty()
        || !global_whitelist.is_empty()
        || !global_blacklist.is_empty()
        || policy.virgin_mode;

    let Some(ip) = ip else {
        if any_rules_present {
            return IpPolicyDecision::Deny {
                reason: "ip_required_for_key",
                message: "Client IP required",
                detail: "API key IP policy requires a resolvable client IP for this request."
                    .to_string(),
            };
        }
        return IpPolicyDecision::Allow;
    };

    if ip_matches_any(&global_blacklist, ip) {
        return IpPolicyDecision::Deny {
            reason: "ip_global_blacklisted",
            message: "IP blocked",
            detail: format!("Client IP '{}' is on the global blacklist.", ip),
        };
    }

    if ip_matches_any(&policy.blacklist, ip) {
        return IpPolicyDecision::Deny {
            reason: "ip_blacklisted",
            message: "IP blocked",
            detail: format!("Client IP '{}' is on this API key's blacklist.", ip),
        };
    }

    if policy.virgin_mode && !policy.virgin_resolved {
        let seen = match record_seen_ip(pool, record.internal_id, ip).await {
            Ok(value) => value,
            Err(err) => {
                error!(error = %err, "failed to record seen ip during virgin learning");
                return IpPolicyDecision::Allow;
            }
        };

        let new_request_count: i32 =
            match increment_virgin_request_count(pool, record.internal_id).await {
                Ok(value) => value,
                Err(err) => {
                    error!(error = %err, "failed to increment virgin request count");
                    policy.virgin_request_count + 1
                }
            };

        let request_threshold_hit: bool = policy.virgin_until_n_requests > 0
            && new_request_count >= policy.virgin_until_n_requests;
        let distinct_threshold_hit: bool = policy.max_whitelist_ips > 0
            && seen.distinct_ip_count as i32 >= policy.max_whitelist_ips;

        if request_threshold_hit || distinct_threshold_hit {
            let max_take: Option<i32> = if policy.max_whitelist_ips > 0 {
                Some(policy.max_whitelist_ips)
            } else {
                None
            };
            if let Err(err) =
                promote_seen_ips_to_whitelist(pool, record.internal_id, max_take).await
            {
                error!(error = %err, "failed to promote seen ips to whitelist");
            }
            if let Err(err) = mark_virgin_resolved(pool, record.internal_id).await {
                error!(error = %err, "failed to mark virgin key as resolved");
            }
            invalidate_ip_policy(record.internal_id);
        }

        return IpPolicyDecision::Allow;
    }

    if !global_whitelist.is_empty() && !ip_matches_any(&global_whitelist, ip) {
        return IpPolicyDecision::Deny {
            reason: "ip_not_in_global_whitelist",
            message: "IP not allowed",
            detail: format!(
                "Client IP '{}' is not on the global whitelist for this deployment.",
                ip
            ),
        };
    }

    if !policy.whitelist.is_empty() && !ip_matches_any(&policy.whitelist, ip) {
        return IpPolicyDecision::Deny {
            reason: "ip_not_whitelisted",
            message: "IP not allowed",
            detail: format!("Client IP '{}' is not on this API key's whitelist.", ip),
        };
    }

    IpPolicyDecision::Allow
}

fn is_gateway_deferred_route(req: &HttpRequest) -> bool {
    matches!(
        req.path(),
        "/gateway/query"
            | "/gateway/data"
            | "/gateway/fetch"
            | "/gateway/update"
            | "/gateway/insert"
            | "/gateway/delete"
    )
}

#[allow(clippy::too_many_arguments)]
fn build_log_context(
    request_id: String,
    presented: Option<&PresentedApiKey>,
    authenticated: bool,
    authorized: bool,
    enforced: bool,
    api_key_id: Option<String>,
    internal_api_key_id: Option<i64>,
    api_key_public_id: Option<String>,
    reason: impl Into<String>,
) -> GatewayAuthOutcome {
    GatewayAuthOutcome {
        request_id,
        log_context: RequestAuthLogContext {
            api_key_id,
            internal_api_key_id,
            presented_api_key_public_id: api_key_public_id
                .or_else(|| presented.and_then(|value| value.public_id.clone())),
            presented_api_key_hash: presented.map(|value| value.fingerprint_hash.clone()),
            presented_api_key_salt: presented.map(|value| value.fingerprint_salt.clone()),
            api_key_authenticated: authenticated,
            api_key_authorized: authorized,
            api_key_enforced: enforced,
            api_key_auth_reason: Some(reason.into()),
        },
        bound_client_name: None,
        response: None,
        force_deferred_queue: false,
        force_deferred_reason: None,
    }
}

fn rejection_response(
    status: StatusCode,
    message: impl Into<String>,
    details: impl Into<String>,
) -> HttpResponse {
    match status {
        StatusCode::FORBIDDEN => forbidden(message, details),
        StatusCode::SERVICE_UNAVAILABLE => service_unavailable(message, details),
        _ => unauthorized(message, details),
    }
}

fn spawn_auth_attempt_log(
    logging_task_limiter: Option<Arc<Semaphore>>,
    pool: Option<PgPool>,
    logging_trust_x_forwarded_for: bool,
    req: &HttpRequest,
    request_id: &str,
    client_name: Option<&str>,
    required_rights: &[String],
    granted_rights: &[String],
    log_context: &RequestAuthLogContext,
) {
    let Some(pool) = pool else {
        return;
    };

    let resolved_ip: Option<String> =
        extract_client_ip(req, logging_trust_x_forwarded_for).map(|ip| ip.to_string());

    let entry: AuthAttemptLogEntry = AuthAttemptLogEntry {
        request_id: request_id.to_string(),
        api_key_id: log_context.api_key_id.clone(),
        internal_api_key_id: log_context.internal_api_key_id,
        api_key_public_id: log_context.presented_api_key_public_id.clone(),
        client_name: client_name.map(str::to_string),
        method: req.method().as_str().to_string(),
        path: req.path().to_string(),
        presented_key_hash: log_context.presented_api_key_hash.clone(),
        presented_key_salt: log_context.presented_api_key_salt.clone(),
        required_rights: required_rights.to_vec(),
        granted_rights: granted_rights.to_vec(),
        authenticated: log_context.api_key_authenticated,
        authorized: log_context.api_key_authorized,
        enforced: log_context.api_key_enforced,
        failure_reason: log_context.api_key_auth_reason.clone(),
        remote_addr: resolved_ip.clone(),
        ipv4_address: resolved_ip,
        user_agent: user_agent(req),
        time: current_epoch_seconds(),
    };

    spawn_best_effort_logging_task(logging_task_limiter, "api_key_auth_log", async move {
        if let Err(err) = insert_auth_attempt_log(&pool, entry).await {
            error!(error = %err, "failed to write api key auth log");
        }
    });
}

fn spawn_last_used_touch(
    logging_task_limiter: Option<Arc<Semaphore>>,
    pool: Option<PgPool>,
    api_key_id: Option<String>,
) {
    let (Some(pool), Some(api_key_id)) = (pool, api_key_id) else {
        return;
    };

    let Ok(api_key_id) = Uuid::parse_str(&api_key_id) else {
        return;
    };

    spawn_best_effort_logging_task(
        logging_task_limiter,
        "api_key_last_used_touch",
        async move {
            if let Err(err) = touch_api_key_last_used(&pool, api_key_id).await {
                error!(error = %err, "failed to update api key last_used_at");
            }
        },
    );
}

pub async fn authorize_gateway_request(
    req: &HttpRequest,
    app_state: &AppState,
    client_name: Option<&str>,
    required_rights: Vec<String>,
) -> GatewayAuthOutcome {
    let request_id: String = Uuid::new_v4().to_string();
    let presented: Option<PresentedApiKey> =
        extract_gateway_api_key(req).map(|value| parse_presented_api_key(&value));
    let auth_pool: Option<sqlx::Pool<sqlx::Postgres>> = auth_store_pool(app_state);
    let logging_task_limiter = app_state.logging_task_limiter.clone();
    let auth_store_configured: bool = app_state.gateway_auth_client_name.is_some();
    let fail_open: bool = app_state
        .gateway_api_key_fail_mode
        .eq_ignore_ascii_case("fail_open");

    let Some(pool) = auth_pool.clone() else {
        let reason: &str = if auth_store_configured {
            "api_key_store_unavailable"
        } else {
            "api_key_store_not_configured"
        };
        if auth_store_configured && fail_open {
            warn!(
                "gateway auth store is configured but unavailable; allowing request without API key enforcement"
            );
        }

        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            presented.as_ref(),
            false,
            fail_open,
            !fail_open,
            None,
            None,
            None,
            if auth_store_configured && fail_open {
                "api_key_store_unavailable_fail_open"
            } else {
                reason
            },
        );
        if auth_store_configured && fail_open && is_gateway_deferred_route(req) {
            outcome.force_deferred_queue = true;
            outcome.force_deferred_reason = Some("api_key_store_unavailable".to_string());
        }
        if auth_store_configured && !fail_open {
            outcome.response = Some(HttpResponse::ServiceUnavailable().json(serde_json::json!({
                "status": "error",
                "code": "auth_store_unavailable",
                "message": "API key validation unavailable",
                "error": "Gateway auth store is configured but currently unreachable."
            })));
        }
        return outcome;
    };

    let enforced: bool = match get_effective_api_key_enforcement_cached(&pool, client_name).await {
        Ok(value) => value,
        Err(err) => {
            error!(error = %err, "failed to load effective api key enforcement");
            if fail_open {
                warn!(
                    error = %err,
                    "api key policy lookup failed; allowing request without API key enforcement (fail_open)"
                );
                let mut outcome: GatewayAuthOutcome = build_log_context(
                    request_id,
                    presented.as_ref(),
                    false,
                    true,
                    false,
                    None,
                    None,
                    None,
                    "api_key_policy_lookup_failed_fail_open",
                );
                if is_gateway_deferred_route(req) {
                    outcome.force_deferred_queue = true;
                    outcome.force_deferred_reason =
                        Some("api_key_policy_lookup_failed".to_string());
                }
                return outcome;
            }
            let mut outcome: GatewayAuthOutcome = build_log_context(
                request_id,
                presented.as_ref(),
                false,
                false,
                false,
                None,
                None,
                None,
                "api_key_policy_lookup_failed",
            );
            outcome.response = Some(rejection_response(
                StatusCode::SERVICE_UNAVAILABLE,
                "API key policy unavailable",
                "Gateway API key enforcement could not be evaluated.",
            ));
            return outcome;
        }
    };

    let required_rights: Vec<String> = if required_rights.is_empty() {
        vec!["gateway.access".to_string()]
    } else {
        required_rights
    };

    let Some(presented) = presented else {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            None,
            false,
            !enforced,
            enforced,
            None,
            None,
            None,
            if enforced {
                "missing_api_key"
            } else {
                "api_key_not_enforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "API key required",
                "Missing API key. Use X-Athena-Key.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &[],
            &outcome.log_context,
        );
        return outcome;
    };

    let Some(public_id) = presented.public_id.as_deref() else {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            None,
            None,
            None,
            if enforced {
                "invalid_api_key_format"
            } else {
                "invalid_api_key_format_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Invalid API key",
                "The API key format is invalid.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &[],
            &outcome.log_context,
        );
        return outcome;
    };

    let Some(secret) = presented.secret.as_deref() else {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            None,
            None,
            Some(public_id.to_string()),
            if enforced {
                "invalid_api_key_format"
            } else {
                "invalid_api_key_format_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Invalid API key",
                "The API key format is invalid.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &[],
            &outcome.log_context,
        );
        return outcome;
    };

    let record: Option<ApiKeySecretRecord> = match get_api_key_by_public_id(&pool, public_id).await
    {
        Ok(value) => value,
        Err(err) => {
            error!(error = %err, "failed to look up api key");
            if fail_open {
                warn!(
                    error = %err,
                    "api key lookup failed; allowing request without API key enforcement (fail_open)"
                );
                let mut outcome: GatewayAuthOutcome = build_log_context(
                    request_id,
                    Some(&presented),
                    false,
                    true,
                    false,
                    None,
                    None,
                    Some(public_id.to_string()),
                    "api_key_lookup_failed_fail_open",
                );
                if is_gateway_deferred_route(req) {
                    outcome.force_deferred_queue = true;
                    outcome.force_deferred_reason = Some("api_key_lookup_failed".to_string());
                }
                spawn_auth_attempt_log(
                    logging_task_limiter.clone(),
                    Some(pool),
                    app_state.logging_trust_x_forwarded_for,
                    req,
                    &outcome.request_id,
                    client_name,
                    &required_rights,
                    &[],
                    &outcome.log_context,
                );
                return outcome;
            }
            let mut outcome: GatewayAuthOutcome = build_log_context(
                request_id,
                Some(&presented),
                false,
                false,
                enforced,
                None,
                None,
                Some(public_id.to_string()),
                "api_key_lookup_failed",
            );
            outcome.response = Some(rejection_response(
                StatusCode::SERVICE_UNAVAILABLE,
                "API key validation unavailable",
                "The API key store could not validate the presented key.",
            ));
            return outcome;
        }
    };

    let Some(record) = record else {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            None,
            None,
            Some(public_id.to_string()),
            if enforced {
                "unknown_api_key"
            } else {
                "unknown_api_key_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Invalid API key",
                "No API key matched the provided identifier.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &[],
            &outcome.log_context,
        );
        return outcome;
    };

    let expected_hash: String = digest(format!("{}:{}", record.key_salt, secret));
    if expected_hash != record.key_hash {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            if enforced {
                "invalid_api_key_secret"
            } else {
                "invalid_api_key_secret_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Invalid API key",
                "The provided API key secret does not match.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        return outcome;
    }

    if !record.record.is_active {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            if enforced {
                "inactive_api_key"
            } else {
                "inactive_api_key_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Inactive API key",
                "The API key has been disabled.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        return outcome;
    }

    if let Some(expires_at) = record.record.expires_at
        && expires_at < Utc::now()
    {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            false,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            if enforced {
                "expired_api_key"
            } else {
                "expired_api_key_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::UNAUTHORIZED,
                "Expired API key",
                "The API key has expired.",
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        return outcome;
    }

    if let Some(bound_client_name) = record.record.client_name.as_deref()
        && client_name.filter(|value| !value.is_empty()) != Some(bound_client_name)
    {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            true,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            if enforced {
                "api_key_client_mismatch"
            } else {
                "api_key_client_mismatch_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::FORBIDDEN,
                "API key client mismatch",
                format!(
                    "The API key is scoped to client '{}', not '{}'.",
                    bound_client_name,
                    client_name.unwrap_or_default()
                ),
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        return outcome;
    }

    let client_ip: Option<IpAddr> =
        extract_client_ip(req, app_state.inbound_rate_limit_trust_x_forwarded_for);

    let missing_rights: Vec<String> =
        missing_required_rights(&record.record.rights, &required_rights);
    if !missing_rights.is_empty() {
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            true,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            if enforced {
                "missing_api_key_rights"
            } else {
                "missing_api_key_rights_unenforced"
            },
        );
        if enforced {
            outcome.response = Some(rejection_response(
                StatusCode::FORBIDDEN,
                "Insufficient API key rights",
                format!("Missing required rights: {}", missing_rights.join(", ")),
            ));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool.clone()),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        spawn_last_used_touch(
            logging_task_limiter.clone(),
            Some(pool),
            Some(record.record.id.clone()),
        );
        return outcome;
    }

    let ip_decision: IpPolicyDecision =
        enforce_ip_policy(&pool, &record, client_name, client_ip).await;
    if let IpPolicyDecision::Deny {
        reason,
        message,
        detail,
    } = ip_decision
    {
        let reason_code: String = if enforced {
            reason.to_string()
        } else {
            format!("{}_unenforced", reason)
        };
        let mut outcome: GatewayAuthOutcome = build_log_context(
            request_id,
            Some(&presented),
            true,
            !enforced,
            enforced,
            Some(record.record.id.clone()),
            Some(record.internal_id),
            Some(record.record.public_id.clone()),
            reason_code,
        );
        if enforced {
            outcome.response = Some(rejection_response(StatusCode::FORBIDDEN, message, detail));
        }
        spawn_auth_attempt_log(
            logging_task_limiter.clone(),
            Some(pool.clone()),
            app_state.logging_trust_x_forwarded_for,
            req,
            &outcome.request_id,
            client_name,
            &required_rights,
            &record.record.rights,
            &outcome.log_context,
        );
        spawn_last_used_touch(
            logging_task_limiter.clone(),
            Some(pool),
            Some(record.record.id.clone()),
        );
        return outcome;
    }

    let mut outcome: GatewayAuthOutcome = build_log_context(
        request_id,
        Some(&presented),
        true,
        true,
        enforced,
        Some(record.record.id.clone()),
        Some(record.internal_id),
        Some(record.record.public_id.clone()),
        if enforced {
            "api_key_authenticated"
        } else {
            "api_key_authenticated_unenforced"
        },
    );
    outcome.bound_client_name = record.record.client_name.clone();
    spawn_auth_attempt_log(
        logging_task_limiter.clone(),
        Some(pool.clone()),
        app_state.logging_trust_x_forwarded_for,
        req,
        &outcome.request_id,
        client_name,
        &required_rights,
        &record.record.rights,
        &outcome.log_context,
    );
    spawn_last_used_touch(logging_task_limiter, Some(pool), Some(record.record.id));
    outcome
}

/// Outcome when either the static admin secret or a gateway API key authorized the request.
#[derive(Debug)]
pub enum AdminOrGatewayAuth {
    StaticAdmin,
    Gateway(GatewayAuthOutcome),
}

/// Require `ATHENA_KEY_12` (via `X-Athena-Key` / `X-Athena-Admin-Key`) **or** a valid gateway
/// `ath_*` key with `required_rights`.
///
/// When the admin env var is unset, only gateway auth is used. When it is set but headers do not
/// match, gateway auth is attempted so the same `X-Athena-Key` can carry an `ath_*` secret.
pub async fn require_admin_or_gateway(
    req: &HttpRequest,
    app_state: &AppState,
    client_name: Option<&str>,
    required_rights: Vec<String>,
) -> Result<AdminOrGatewayAuth, HttpResponse> {
    match static_admin_key_status(req) {
        StaticAdminKeyStatus::Authorized => Ok(AdminOrGatewayAuth::StaticAdmin),
        StaticAdminKeyStatus::NotConfigured => {
            if app_state.gateway_auth_client_name.is_none() {
                return Err(unauthorized(
                    "Authentication required",
                    "Set ATHENA_KEY_12 or configure gateway.api_key_client (API key store).",
                ));
            }
            let outcome: GatewayAuthOutcome =
                authorize_gateway_request(req, app_state, client_name, required_rights).await;
            if let Some(resp) = outcome.response {
                return Err(resp);
            }
            Ok(AdminOrGatewayAuth::Gateway(outcome))
        }
        StaticAdminKeyStatus::NotAuthorized => {
            if app_state.gateway_auth_client_name.is_none() {
                return Err(unauthorized(
                    "Authentication required",
                    "Invalid or missing API key. Use X-Athena-Key with ATHENA_KEY_12 or configure gateway.api_key_client.",
                ));
            }
            let outcome: GatewayAuthOutcome =
                authorize_gateway_request(req, app_state, client_name, required_rights).await;
            if let Some(resp) = outcome.response {
                return Err(resp);
            }
            Ok(AdminOrGatewayAuth::Gateway(outcome))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        delete_right_for_resource, parse_presented_api_key, query_right, read_right_for_resource,
        right_matches, rpc_right, write_right_for_resource,
    };
    use crate::api::gateway::auth::PresentedApiKey;

    #[test]
    fn api_key_parser_extracts_identifier_and_secret() {
        let parsed: PresentedApiKey = parse_presented_api_key("ath_public123.secret456");
        assert_eq!(parsed.public_id.as_deref(), Some("public123"));
        assert_eq!(parsed.secret.as_deref(), Some("secret456"));
        assert!(!parsed.fingerprint_hash.is_empty());
        assert!(!parsed.fingerprint_salt.is_empty());
    }

    #[test]
    fn wildcard_rights_match_expected_shapes() {
        assert!(right_matches("users.read", "users.read"));
        assert!(right_matches("users.*", "users.read"));
        assert!(right_matches("*.read", "users.read"));
        assert!(right_matches("gateway.read", "users.read"));
        assert!(right_matches("gateway.*", "users.delete"));
        assert!(right_matches("*", "users.read"));
        assert!(!right_matches("users.write", "users.read"));
        assert!(!right_matches("tickets.read", "users.read"));
    }

    #[test]
    fn resource_right_helpers_fall_back_to_gateway_scopes() {
        assert_eq!(read_right_for_resource(Some("users")), "users.read");
        assert_eq!(write_right_for_resource(Some("users")), "users.write");
        assert_eq!(delete_right_for_resource(Some("users")), "users.delete");
        assert_eq!(
            read_right_for_resource(Some("public.users")),
            "gateway.read"
        );
        assert_eq!(query_right(), "gateway.query");
        assert_eq!(rpc_right(), "gateway.rpc.execute");
    }
}