//! TLS support for the carapace gateway
//!
//! Provides:
//! - Self-signed certificate auto-generation using `rcgen`
//! - Certificate and key loading from PEM files
//! - SHA-256 fingerprint computation for trust-on-first-use pairing
//! - TLS configuration types for the gateway config schema
//! - Cluster CA management for mTLS gateway-to-gateway authentication
//! - mTLS configuration and rustls server/client config builders

pub mod ca;

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use chrono::Datelike;
use rustls::client::danger::HandshakeSignatureValid;
use rustls::server::danger::{ClientCertVerified, ClientCertVerifier};
use rustls_pki_types::pem::PemObject;
use rustls_pki_types::{CertificateDer, PrivateKeyDer, UnixTime};
use sha2::{Digest, Sha256};
use thiserror::Error;
use tracing::{debug, info, warn};

/// Errors that can occur during TLS setup
#[derive(Error, Debug)]
pub enum TlsError {
    #[error("Failed to generate self-signed certificate: {0}")]
    CertGenerationFailed(String),

    #[error("Failed to read certificate file {path}: {message}")]
    CertReadError { path: String, message: String },

    #[error("Failed to read key file {path}: {message}")]
    KeyReadError { path: String, message: String },

    #[error("No certificates found in PEM file: {0}")]
    NoCertsFound(String),

    #[error("No private key found in PEM file: {0}")]
    NoKeyFound(String),

    #[error("Failed to build TLS config: {0}")]
    ConfigBuildError(String),

    #[error("Failed to create TLS directory {path}: {message}")]
    DirCreationError { path: String, message: String },

    #[error("Failed to write certificate file {path}: {message}")]
    CertWriteError { path: String, message: String },

    #[error("Failed to write key file {path}: {message}")]
    KeyWriteError { path: String, message: String },
}

/// TLS configuration parsed from the gateway config
#[derive(Debug, Clone)]
pub struct TlsConfig {
    /// Whether TLS is enabled
    pub enabled: bool,
    /// Path to the certificate PEM file (optional; if not provided, auto-generate)
    pub cert_path: Option<PathBuf>,
    /// Path to the private key PEM file (optional; if not provided, auto-generate)
    pub key_path: Option<PathBuf>,
    /// Whether to auto-generate a self-signed certificate when cert/key paths are not provided
    pub auto_generate: bool,
}

impl Default for TlsConfig {
    fn default() -> Self {
        TlsConfig {
            enabled: false,
            cert_path: None,
            key_path: None,
            auto_generate: true,
        }
    }
}

/// Parse TLS configuration from the loaded JSON config value.
///
/// Looks for `gateway.tls` object with keys:
/// - `enabled` (bool, default false)
/// - `certPath` (string, optional)
/// - `keyPath` (string, optional)
/// - `autoGenerate` (bool, default true)
pub fn parse_tls_config(cfg: &serde_json::Value) -> TlsConfig {
    let tls_obj = cfg
        .get("gateway")
        .and_then(|g| g.get("tls"))
        .and_then(|t| t.as_object());

    let tls_obj = match tls_obj {
        Some(obj) => obj,
        None => return TlsConfig::default(),
    };

    let enabled = tls_obj
        .get("enabled")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);

    let cert_path = tls_obj
        .get("certPath")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let key_path = tls_obj
        .get("keyPath")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let auto_generate = tls_obj
        .get("autoGenerate")
        .and_then(|v| v.as_bool())
        .unwrap_or(true);

    TlsConfig {
        enabled,
        cert_path,
        key_path,
        auto_generate,
    }
}

/// Result of TLS setup, containing the rustls ServerConfig and certificate fingerprint
pub struct TlsSetupResult {
    /// The rustls server configuration, ready to use with axum-server
    pub server_config: Arc<rustls::ServerConfig>,
    /// SHA-256 fingerprint of the server certificate (hex-encoded, colon-separated)
    pub fingerprint: String,
    /// Path to the certificate file being used
    pub cert_path: PathBuf,
    /// Path to the key file being used
    pub key_path: PathBuf,
}

/// Ensure a self-signed certificate exists in the default TLS directory,
/// generating one if necessary.  Returns the cert and key paths.
fn auto_generate_certificate() -> Result<(PathBuf, PathBuf), TlsError> {
    info!("Auto-generating self-signed TLS certificate");
    let tls_dir = default_tls_dir();
    ensure_tls_dir(&tls_dir)?;

    let cert_path = tls_dir.join("cert.pem");
    let key_path = tls_dir.join("key.pem");

    if !cert_path.exists() || !key_path.exists() {
        generate_self_signed_cert(&cert_path, &key_path)?;
        info!(
            "Generated self-signed certificate at {}",
            cert_path.display()
        );
    } else {
        info!(
            "Using existing self-signed certificate at {}",
            cert_path.display()
        );
    }

    Ok((cert_path, key_path))
}

/// Resolve the certificate and key file paths.
///
/// When both paths are provided in the config they are returned as-is.
/// When neither is provided and `auto_generate` is enabled, the default TLS
/// directory is created and a self-signed certificate is generated (if not
/// already present).  Returns an error for any other combination.
fn resolve_certificate_paths(config: &TlsConfig) -> Result<(PathBuf, PathBuf), TlsError> {
    match (&config.cert_path, &config.key_path) {
        (Some(cert), Some(key)) => {
            info!("Loading TLS certificate from provided paths");
            Ok((cert.clone(), key.clone()))
        }
        (None, None) if config.auto_generate => auto_generate_certificate(),
        _ => Err(TlsError::ConfigBuildError(
            "TLS enabled but cert/key paths are incomplete and auto-generate is disabled"
                .to_string(),
        )),
    }
}

/// Load certificates and key, compute the fingerprint, and assemble a
/// `rustls::ServerConfig`.
fn build_server_config(
    cert_path: &Path,
    key_path: &Path,
) -> Result<(Arc<rustls::ServerConfig>, String), TlsError> {
    let certs = load_certs(cert_path)?;
    let key = load_private_key(key_path)?;

    // Compute fingerprint from the first (leaf) certificate
    let fingerprint = compute_cert_fingerprint(&certs[0]);

    // Ensure a crypto provider is installed (required by rustls 0.23+)
    // This is idempotent; if already installed, the Err is ignored.
    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();

    let server_config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(certs, key)
        .map_err(|e| TlsError::ConfigBuildError(e.to_string()))?;

    Ok((Arc::new(server_config), fingerprint))
}

/// Set up TLS based on the provided configuration.
///
/// If `cert_path` and `key_path` are provided, loads them from disk.
/// If they are not provided and `auto_generate` is true, generates a self-signed
/// certificate and stores it in the default TLS directory.
///
/// Returns a `TlsSetupResult` containing the rustls ServerConfig and fingerprint.
pub fn setup_tls(config: &TlsConfig) -> Result<TlsSetupResult, TlsError> {
    let (cert_path, key_path) = resolve_certificate_paths(config)?;
    let (server_config, fingerprint) = build_server_config(&cert_path, &key_path)?;

    Ok(TlsSetupResult {
        server_config,
        fingerprint,
        cert_path,
        key_path,
    })
}

/// Get the default TLS directory path.
///
/// Uses `~/.config/carapace/tls/` if a home directory is available,
/// otherwise falls back to `.carapace/tls/` in the current directory.
pub fn default_tls_dir() -> PathBuf {
    dirs::config_dir()
        .unwrap_or_else(|| PathBuf::from(".config"))
        .join("carapace")
        .join("tls")
}

/// Ensure the TLS directory exists, creating it if necessary
fn ensure_tls_dir(dir: &Path) -> Result<(), TlsError> {
    if !dir.exists() {
        std::fs::create_dir_all(dir).map_err(|e| TlsError::DirCreationError {
            path: dir.display().to_string(),
            message: e.to_string(),
        })?;
        debug!("Created TLS directory: {}", dir.display());
    }
    // Ratchet the TLS dir to 0o700 (owner-only). The dir holds the
    // daemon's private key; world-readable mode lets local users
    // observe filenames and creation timestamps, and on a shared
    // workstation invites symlink-plant races against the predictable
    // `key.pem` / `cert.pem` paths. Best-effort warn on failure
    // (filesystems without Unix permissions, mounted with weird
    // options, etc.) — symlink defense at the open-syscall layer
    // remains the load-bearing guard.
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let perms = std::fs::Permissions::from_mode(0o700);
        if let Err(e) = std::fs::set_permissions(dir, perms) {
            warn!(
                path = %dir.display(),
                error = %e,
                "failed to chmod 0o700 on TLS directory; symlink defense at open(2) \
                 remains the load-bearing guard"
            );
        }
    }
    Ok(())
}

/// Generate a self-signed certificate and private key, writing them to PEM files.
///
/// The certificate is valid for "localhost", "127.0.0.1", and "::1" as Subject
/// Alternative Names, making it suitable for local development and loopback access.
pub fn generate_self_signed_cert(cert_path: &Path, key_path: &Path) -> Result<(), TlsError> {
    use rcgen::{CertificateParams, KeyPair, SanType};

    let mut params = CertificateParams::new(vec!["localhost".to_string()])
        .map_err(|e| TlsError::CertGenerationFailed(e.to_string()))?;

    // Add Subject Alternative Names for common local addresses
    params.subject_alt_names = vec![
        SanType::DnsName(
            "localhost"
                .try_into()
                .map_err(|e: rcgen::Error| TlsError::CertGenerationFailed(e.to_string()))?,
        ),
        SanType::IpAddress(std::net::IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1))),
        SanType::IpAddress(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)),
    ];

    // Set a reasonable validity period (365 days)
    let now = chrono::Utc::now();
    params.not_before = rcgen::date_time_ymd(now.year(), now.month() as u8, now.day() as u8);
    let future = now + chrono::Duration::days(365);
    params.not_after =
        rcgen::date_time_ymd(future.year(), future.month() as u8, future.day() as u8);

    // Set distinguished name
    params
        .distinguished_name
        .push(rcgen::DnType::CommonName, "Carapace Gateway");
    params
        .distinguished_name
        .push(rcgen::DnType::OrganizationName, "Carapace");

    // Generate key pair and certificate
    let key_pair =
        KeyPair::generate().map_err(|e| TlsError::CertGenerationFailed(e.to_string()))?;

    let cert = params
        .self_signed(&key_pair)
        .map_err(|e| TlsError::CertGenerationFailed(e.to_string()))?;

    // Write certificate PEM via atomic tmp + O_NOFOLLOW + O_EXCL +
    // 0o600 + rename. SECURITY: the prior `std::fs::write(cert_path, ...)`
    // followed symlinks and used the default umask. A same-uid
    // attacker who pre-planted a symlink at `cert_path` would have
    // the daemon write the cert to the symlink's target. The cert
    // itself is public, so disclosure is not a concern; the issue is
    // that the redirect can be used to plant attacker-chosen content
    // at an attacker-chosen path that the daemon's identity can write
    // (the trust-on-first-use for any later reader of that path).
    let cert_pem = cert.pem();
    write_tls_pem_atomic_owner_only(cert_path, cert_pem.as_bytes()).map_err(|e| {
        TlsError::CertWriteError {
            path: cert_path.display().to_string(),
            message: e.to_string(),
        }
    })?;

    // Write private key PEM via atomic tmp + O_NOFOLLOW + O_EXCL +
    // 0o600 + rename. SECURITY: the prior path was the higher-stakes
    // half of the same defect. The pre-fix sequence was:
    //   1. `std::fs::write(key_path, key_pem.as_bytes())` — follows
    //      symlinks, default umask 0o644.
    //   2. `std::fs::set_permissions(key_path, 0o600)` — also path-
    //      based, also follows symlinks.
    // A same-uid attacker who pre-planted a symlink at `key_path`
    // captured a freshly-generated private key (information
    // disclosure on the daemon's TLS identity) AND had the daemon
    // chmod the symlink target to 0o600 — usable as a primitive to
    // restrict access on any path the daemon's identity can chmod.
    // The `create_atomic_tmp_owner_only` helper opens the tmp with
    // `O_NOFOLLOW | O_EXCL | O_WRONLY` at mode 0o600 in a single
    // syscall, then `rename(tmp, dst)` atomically replaces any
    // existing dirent at `key_path` — including a symlink — without
    // ever following it.
    let key_pem = key_pair.serialize_pem();
    write_tls_pem_atomic_owner_only(key_path, key_pem.as_bytes()).map_err(|e| {
        TlsError::KeyWriteError {
            path: key_path.display().to_string(),
            message: e.to_string(),
        }
    })?;

    Ok(())
}

/// Write a TLS PEM payload to `path` via an atomic tmp + rename
/// pipeline that refuses to follow symlinks and forces 0o600 at
/// creation time. See `generate_self_signed_cert` for the threat
/// model that motivated this helper.
pub(super) fn write_tls_pem_atomic_owner_only(path: &Path, pem: &[u8]) -> std::io::Result<()> {
    use std::io::Write;
    let tmp_path = crate::paths::atomic_tmp_path(path, "tlspem");
    let write_result = (|| -> std::io::Result<()> {
        let mut file = crate::paths::create_atomic_tmp_owner_only(&tmp_path)?;
        file.write_all(pem)?;
        file.sync_all()?;
        drop(file);
        std::fs::rename(&tmp_path, path)?;
        crate::paths::sync_parent_dir_best_effort_blocking(path);
        Ok(())
    })();
    if write_result.is_err() {
        let _ = std::fs::remove_file(&tmp_path);
    }
    write_result
}

/// Load certificates from an in-memory PEM slice. SECURITY: prefer
/// this over `load_certs(path)` whenever the bytes have already
/// been loaded through `paths::read_to_vec_no_hang_no_follow_capped`
/// — re-opening the path via `pem_file_iter` would re-introduce
/// the FIFO/symlink/oversize attack surface the hardened read
/// just closed. Used by `ca::load` (B137) to avoid the
/// hardened-read-then-rustls-reread TOCTOU window.
pub(super) fn load_certs_from_slice(
    bytes: &[u8],
    source_label: &str,
) -> Result<Vec<CertificateDer<'static>>, TlsError> {
    let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_slice_iter(bytes)
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| TlsError::CertReadError {
            path: source_label.to_string(),
            message: e.to_string(),
        })?;
    if certs.is_empty() {
        return Err(TlsError::NoCertsFound(source_label.to_string()));
    }
    Ok(certs)
}

/// Load certificates from a PEM file
pub fn load_certs(path: &Path) -> Result<Vec<CertificateDer<'static>>, TlsError> {
    let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_file_iter(path)
        .map_err(|e| TlsError::CertReadError {
            path: path.display().to_string(),
            message: e.to_string(),
        })?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| TlsError::CertReadError {
            path: path.display().to_string(),
            message: e.to_string(),
        })?;

    if certs.is_empty() {
        return Err(TlsError::NoCertsFound(path.display().to_string()));
    }

    debug!(
        "Loaded {} certificate(s) from {}",
        certs.len(),
        path.display()
    );
    Ok(certs)
}

/// Load a private key from a PEM file.
///
/// Supports PKCS#8 and RSA/EC keys.
pub fn load_private_key(path: &Path) -> Result<PrivateKeyDer<'static>, TlsError> {
    match PrivateKeyDer::from_pem_file(path) {
        Ok(key) => {
            debug!("Loaded private key from {}", path.display());
            Ok(key)
        }
        Err(rustls_pki_types::pem::Error::NoItemsFound) => {
            Err(TlsError::NoKeyFound(path.display().to_string()))
        }
        Err(e) => Err(TlsError::KeyReadError {
            path: path.display().to_string(),
            message: e.to_string(),
        }),
    }
}

/// Compute the SHA-256 fingerprint of a DER-encoded certificate.
///
/// Returns the fingerprint as a colon-separated hex string, e.g.:
/// `AB:CD:EF:01:23:45:...`
///
/// This fingerprint is used for trust-on-first-use (TOFU) pairing,
/// where nodes verify the gateway's identity by comparing fingerprints.
pub fn compute_cert_fingerprint(cert_der: &CertificateDer<'_>) -> String {
    let mut hasher = Sha256::new();
    hasher.update(cert_der.as_ref());
    let hash = hasher.finalize();

    hash.iter()
        .map(|b| format!("{:02X}", b))
        .collect::<Vec<_>>()
        .join(":")
}

// ============================================================================
// mTLS configuration
// ============================================================================

/// mTLS configuration for gateway-to-gateway communication.
#[derive(Debug, Clone)]
pub struct MtlsConfig {
    /// Whether mTLS is enabled for gateway connections.
    pub enabled: bool,
    /// Path to the cluster CA certificate PEM file.
    pub ca_cert: Option<PathBuf>,
    /// Optional path to a CRL file for revocation checks.
    pub crl_path: Option<PathBuf>,
    /// Path to this node's certificate PEM file.
    pub node_cert: Option<PathBuf>,
    /// Path to this node's private key PEM file.
    pub node_key: Option<PathBuf>,
    /// Whether to require client certificates from connecting gateways.
    pub require_client_cert: bool,
}

impl Default for MtlsConfig {
    fn default() -> Self {
        Self {
            enabled: false,
            ca_cert: None,
            crl_path: None,
            node_cert: None,
            node_key: None,
            require_client_cert: true,
        }
    }
}

/// Parse mTLS configuration from the loaded JSON config value.
///
/// Looks for `gateway.mtls` object with keys:
/// - `enabled` (bool, default false)
/// - `caCert` (string, path to CA certificate PEM)
/// - `crlPath` (string, optional path to CRL JSON file)
/// - `nodeCert` (string, path to node certificate PEM)
/// - `nodeKey` (string, path to node private key PEM)
/// - `requireClientCert` (bool, default true)
pub fn parse_mtls_config(cfg: &serde_json::Value) -> MtlsConfig {
    let mtls_obj = cfg
        .get("gateway")
        .and_then(|g| g.get("mtls"))
        .and_then(|t| t.as_object());

    let mtls_obj = match mtls_obj {
        Some(obj) => obj,
        None => return MtlsConfig::default(),
    };

    let enabled = mtls_obj
        .get("enabled")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);

    let ca_cert = mtls_obj
        .get("caCert")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let crl_path = mtls_obj
        .get("crlPath")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let node_cert = mtls_obj
        .get("nodeCert")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let node_key = mtls_obj
        .get("nodeKey")
        .and_then(|v| v.as_str())
        .map(PathBuf::from);

    let require_client_cert = mtls_obj
        .get("requireClientCert")
        .and_then(|v| v.as_bool())
        .unwrap_or(true);

    MtlsConfig {
        enabled,
        ca_cert,
        crl_path,
        node_cert,
        node_key,
        require_client_cert,
    }
}

/// Result of mTLS setup, containing server and client TLS configurations.
pub struct MtlsSetupResult {
    /// Server-side TLS config with client certificate verification.
    pub server_config: Arc<rustls::ServerConfig>,
    /// Client-side TLS config that presents this node's certificate.
    pub client_config: Arc<rustls::ClientConfig>,
    /// SHA-256 fingerprint of this node's certificate.
    pub node_fingerprint: String,
    /// SHA-256 fingerprint of the CA certificate.
    pub ca_fingerprint: String,
}

#[derive(Debug)]
struct CrlClientCertVerifier {
    inner: Arc<dyn ClientCertVerifier>,
    revoked: HashSet<String>,
}

impl CrlClientCertVerifier {
    fn new(inner: Arc<dyn ClientCertVerifier>, revoked: HashSet<String>) -> Self {
        Self { inner, revoked }
    }

    fn is_revoked(&self, cert: &CertificateDer<'_>) -> bool {
        if self.revoked.is_empty() {
            return false;
        }
        let fingerprint = compute_cert_fingerprint(cert).to_ascii_lowercase();
        self.revoked.contains(&fingerprint)
    }
}

impl ClientCertVerifier for CrlClientCertVerifier {
    fn offer_client_auth(&self) -> bool {
        self.inner.offer_client_auth()
    }

    fn client_auth_mandatory(&self) -> bool {
        self.inner.client_auth_mandatory()
    }

    fn root_hint_subjects(&self) -> &[rustls::DistinguishedName] {
        self.inner.root_hint_subjects()
    }

    fn verify_client_cert(
        &self,
        end_entity: &CertificateDer<'_>,
        intermediates: &[CertificateDer<'_>],
        now: UnixTime,
    ) -> Result<ClientCertVerified, rustls::Error> {
        let verified = self
            .inner
            .verify_client_cert(end_entity, intermediates, now)?;
        if self.is_revoked(end_entity) {
            return Err(rustls::Error::InvalidCertificate(
                rustls::CertificateError::Revoked,
            ));
        }
        Ok(verified)
    }

    fn verify_tls12_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &rustls::DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        self.inner.verify_tls12_signature(message, cert, dss)
    }

    fn verify_tls13_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &rustls::DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        self.inner.verify_tls13_signature(message, cert, dss)
    }

    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        self.inner.supported_verify_schemes()
    }

    fn requires_raw_public_keys(&self) -> bool {
        self.inner.requires_raw_public_keys()
    }
}

/// Load CRL fingerprints from disk, FAILING CLOSED on read or parse
/// errors. The prior implementation silently swallowed every error
/// path (missing file, read error, parse error) into an empty
/// `HashSet`, which the verifier then treats as "no certs revoked" —
/// every previously-revoked client cert would be silently accepted
/// after the CRL became unreadable.
///
/// Cap on the CRL file size when loaded at TLS-config setup or
/// hot-reload. 16 MiB is generously above any realistic CRL
/// (~250,000 fingerprint entries); the cap refuses a self-DoS
/// from a misconfigured path pointing at `/dev/zero` or a
/// runaway log file.
const CRL_FILE_MAX_BYTES: u64 = 16 * 1024 * 1024;

/// Semantics:
/// - `crl_path == None`: legitimately no CRL configured — Ok(empty).
/// - `crl_path = Some(path)` but file does not exist:
///   - if `warn_on_missing` (operator explicitly configured a path):
///     warn AND fail with `CrlUnavailable` so the mTLS bring-up
///     refuses rather than silently accept revoked certs.
///   - otherwise (implicit path derived from ca_dir): Ok(empty).
/// - File exists but read/parse fails: warn AND fail.
fn load_crl_fingerprints(
    crl_path: Option<&Path>,
    warn_on_missing: bool,
) -> Result<HashSet<String>, TlsError> {
    let Some(path) = crl_path else {
        return Ok(HashSet::new());
    };

    if !path.exists() {
        if warn_on_missing {
            warn!(
                path = %path.display(),
                "configured CRL file not found; refusing to start mTLS with empty revocation list"
            );
            return Err(TlsError::ConfigBuildError(format!(
                "configured CRL file not found: {}",
                path.display()
            )));
        }
        return Ok(HashSet::new());
    }

    // SECURITY: route the CRL load through the capped, no-symlink-
    // follow, no-FIFO-hang helper. Without a cap, a misconfigured
    // path (operator points `gateway.mtls.crlPath` at `/dev/zero`
    // or a runaway log file) OOMs the daemon at startup AND at
    // every TLS-config hot-reload — a self-inflicted DoS where
    // the only fix is to manually edit config. 16 MiB matches the
    // plugin-manifest cap and is generously above any realistic
    // CRL size (a 16 MiB CRL would hold ~250k fingerprints).
    let content_bytes = crate::paths::read_to_vec_no_hang_no_follow_capped(
        path,
        CRL_FILE_MAX_BYTES,
    )
    .map_err(|e| {
        warn!(path = %path.display(), error = %e, "failed to read CRL file");
        TlsError::ConfigBuildError(format!("failed to read CRL file {}: {e}", path.display()))
    })?;
    let Some(content_bytes) = content_bytes else {
        // File reported present above but disappeared between
        // exists() and the open. Treat as missing per the
        // warn_on_missing contract that exists() handled above.
        return Ok(HashSet::new());
    };

    let crl: ca::CertRevocationList = serde_json::from_slice(&content_bytes).map_err(|e| {
        warn!(path = %path.display(), error = %e, "failed to parse CRL file");
        TlsError::ConfigBuildError(format!("failed to parse CRL file {}: {e}", path.display()))
    })?;

    Ok(crl
        .entries
        .into_iter()
        .map(|entry| entry.fingerprint.to_ascii_lowercase())
        .collect())
}

/// Set up mTLS for gateway-to-gateway communication.
///
/// Builds both a `rustls::ServerConfig` (for accepting connections from other
/// gateways, verifying their client certificates against the cluster CA) and
/// a `rustls::ClientConfig` (for connecting to other gateways, presenting
/// this node's certificate).
pub fn setup_mtls(config: &MtlsConfig) -> Result<MtlsSetupResult, TlsError> {
    let ca_cert_path = config.ca_cert.as_deref().ok_or_else(|| {
        TlsError::ConfigBuildError("mTLS enabled but gateway.mtls.caCert is not set".to_string())
    })?;
    let node_cert_path = config.node_cert.as_deref().ok_or_else(|| {
        TlsError::ConfigBuildError("mTLS enabled but gateway.mtls.nodeCert is not set".to_string())
    })?;
    let node_key_path = config.node_key.as_deref().ok_or_else(|| {
        TlsError::ConfigBuildError("mTLS enabled but gateway.mtls.nodeKey is not set".to_string())
    })?;

    // Load CA certificate
    let ca_certs = load_certs(ca_cert_path)?;
    let ca_fingerprint = compute_cert_fingerprint(&ca_certs[0]);

    // Load node certificate and key
    let node_certs = load_certs(node_cert_path)?;
    let node_key = load_private_key(node_key_path)?;
    let node_fingerprint = compute_cert_fingerprint(&node_certs[0]);

    // Ensure crypto provider is installed
    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();

    // Build the root cert store with the cluster CA
    let mut root_store = rustls::RootCertStore::empty();
    for ca_cert in &ca_certs {
        root_store
            .add(ca_cert.clone())
            .map_err(|e| TlsError::ConfigBuildError(format!("failed to add CA cert: {}", e)))?;
    }

    let crl_path = config
        .crl_path
        .clone()
        .or_else(|| ca_cert_path.parent().map(|dir| dir.join(ca::CRL_FILENAME)));
    let revoked = load_crl_fingerprints(crl_path.as_deref(), config.crl_path.is_some())?;

    // -- Server config: verify client certs against the cluster CA --
    let client_verifier = if config.require_client_cert {
        rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store.clone()))
            .build()
            .map_err(|e| {
                TlsError::ConfigBuildError(format!("failed to build client verifier: {}", e))
            })?
    } else {
        rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store.clone()))
            .allow_unauthenticated()
            .build()
            .map_err(|e| {
                TlsError::ConfigBuildError(format!("failed to build client verifier: {}", e))
            })?
    };

    let client_verifier = Arc::new(CrlClientCertVerifier::new(client_verifier, revoked));

    let server_config = rustls::ServerConfig::builder()
        .with_client_cert_verifier(client_verifier)
        .with_single_cert(node_certs.clone(), node_key.clone_key())
        .map_err(|e| TlsError::ConfigBuildError(format!("server config: {}", e)))?;

    // -- Client config: present node cert, verify server against CA --
    let client_config = rustls::ClientConfig::builder()
        .with_root_certificates(root_store)
        .with_client_auth_cert(node_certs, node_key)
        .map_err(|e| TlsError::ConfigBuildError(format!("client config: {}", e)))?;

    info!(
        node_fingerprint = %node_fingerprint,
        ca_fingerprint = %ca_fingerprint,
        require_client_cert = config.require_client_cert,
        "mTLS configured for gateway-to-gateway communication"
    );

    Ok(MtlsSetupResult {
        server_config: Arc::new(server_config),
        client_config: Arc::new(client_config),
        node_fingerprint,
        ca_fingerprint,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    /// Pin the Batch 22 CRL fail-closed contract: when the operator
    /// explicitly set a `crl_path` (warn_on_missing=true) but the
    /// file does not exist, the loader must HARD-FAIL with
    /// `ConfigBuildError` so mTLS bring-up refuses to start with an
    /// empty revocation list. The prior implementation silently
    /// returned an empty `HashSet`, which the verifier then treated
    /// as "no certs revoked" — fail-OPEN.
    #[test]
    fn test_load_crl_fingerprints_explicit_missing_path_hard_errors() {
        let dir = TempDir::new().unwrap();
        let missing = dir.path().join("nonexistent-crl.json");
        let err = load_crl_fingerprints(Some(&missing), /* warn_on_missing */ true)
            .expect_err("explicit-but-missing CRL path must hard-error");
        let TlsError::ConfigBuildError(msg) = err else {
            panic!("expected ConfigBuildError; got {err:?}");
        };
        assert!(
            msg.contains("CRL file not found"),
            "error must name the failure mode; got: {msg}"
        );
        assert!(
            msg.contains("nonexistent-crl.json"),
            "error must include the path so operator can fix it; got: {msg}"
        );
    }

    /// Pin the implicit-derived-path behavior: when the path was
    /// derived (e.g., from `ca_dir/CRL_FILENAME`) and the file does
    /// not exist, the loader returns Ok(empty) — operator did not
    /// explicitly configure a CRL, so absence is "no revocations"
    /// rather than a misconfig signal.
    #[test]
    fn test_load_crl_fingerprints_implicit_missing_path_returns_empty() {
        let dir = TempDir::new().unwrap();
        let missing = dir.path().join("nonexistent-implicit-crl.json");
        let result = load_crl_fingerprints(Some(&missing), /* warn_on_missing */ false)
            .expect("implicit-derived missing path must be Ok(empty)");
        assert!(
            result.is_empty(),
            "implicit missing path must yield empty set"
        );
    }

    /// Pin the `crl_path = None` legitimate-no-CRL path returns
    /// Ok(empty).
    #[test]
    fn test_load_crl_fingerprints_none_path_returns_empty() {
        let result = load_crl_fingerprints(None, false).expect("None path must be Ok(empty)");
        assert!(result.is_empty());
    }

    /// Pin file-exists-but-unparseable hard-errors. A corrupted CRL
    /// file must not silently degrade to empty (which would re-open
    /// the fail-open behavior the Batch 22 fix closed).
    #[test]
    fn test_load_crl_fingerprints_unparseable_hard_errors() {
        let dir = TempDir::new().unwrap();
        let crl_path = dir.path().join("crl.json");
        std::fs::write(&crl_path, b"not valid json").expect("test setup: write corrupted CRL");
        let err = load_crl_fingerprints(Some(&crl_path), true)
            .expect_err("unparseable CRL must hard-error regardless of warn_on_missing");
        let TlsError::ConfigBuildError(msg) = err else {
            panic!("expected ConfigBuildError; got {err:?}");
        };
        assert!(
            msg.contains("parse"),
            "error must name parse failure; got: {msg}"
        );
    }

    #[test]
    fn test_default_tls_config() {
        let config = TlsConfig::default();
        assert!(!config.enabled);
        assert!(config.cert_path.is_none());
        assert!(config.key_path.is_none());
        assert!(config.auto_generate);
    }

    #[test]
    fn test_parse_tls_config_empty() {
        let cfg = serde_json::json!({});
        let tls = parse_tls_config(&cfg);
        assert!(!tls.enabled);
        assert!(tls.cert_path.is_none());
        assert!(tls.key_path.is_none());
        assert!(tls.auto_generate);
    }

    #[test]
    fn test_parse_tls_config_enabled() {
        let cfg = serde_json::json!({
            "gateway": {
                "tls": {
                    "enabled": true,
                    "certPath": "/path/to/cert.pem",
                    "keyPath": "/path/to/key.pem",
                    "autoGenerate": false
                }
            }
        });
        let tls = parse_tls_config(&cfg);
        assert!(tls.enabled);
        assert_eq!(tls.cert_path, Some(PathBuf::from("/path/to/cert.pem")));
        assert_eq!(tls.key_path, Some(PathBuf::from("/path/to/key.pem")));
        assert!(!tls.auto_generate);
    }

    #[test]
    fn test_parse_tls_config_auto_generate_default() {
        let cfg = serde_json::json!({
            "gateway": {
                "tls": {
                    "enabled": true
                }
            }
        });
        let tls = parse_tls_config(&cfg);
        assert!(tls.enabled);
        assert!(tls.cert_path.is_none());
        assert!(tls.key_path.is_none());
        assert!(tls.auto_generate); // default is true
    }

    #[test]
    fn test_parse_tls_config_no_gateway() {
        let cfg = serde_json::json!({
            "other": "stuff"
        });
        let tls = parse_tls_config(&cfg);
        assert!(!tls.enabled);
    }

    #[test]
    fn test_generate_self_signed_cert() {
        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        // Verify files were created
        assert!(cert_path.exists());
        assert!(key_path.exists());

        // Verify cert content looks like PEM
        let cert_content = std::fs::read_to_string(&cert_path).unwrap();
        assert!(cert_content.contains("BEGIN CERTIFICATE"));
        assert!(cert_content.contains("END CERTIFICATE"));

        // Verify key content looks like PEM
        let key_content = std::fs::read_to_string(&key_path).unwrap();
        assert!(key_content.contains("BEGIN PRIVATE KEY"));
        assert!(key_content.contains("END PRIVATE KEY"));
    }

    #[cfg(unix)]
    #[test]
    fn test_generate_self_signed_cert_does_not_follow_planted_key_symlink() {
        // SECURITY regression: a same-uid attacker pre-plants a
        // symlink at the daemon's key path. Before B116 the daemon
        // would write the freshly-generated private key to the
        // symlink's target (information disclosure) and then chmod
        // the redirected target to 0o600 (filesystem primitive on
        // any path the daemon's identity can chmod). After the fix,
        // the atomic-tmp+rename pipeline opens the tmp with
        // O_NOFOLLOW + O_EXCL + 0o600 and `rename` atomically
        // replaces the symlink dirent without ever following it.
        use std::os::unix::fs::{symlink, PermissionsExt};

        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");
        let attacker_target = dir.path().join("attacker-target");
        std::fs::write(&attacker_target, b"untouched").unwrap();
        symlink(&attacker_target, &key_path).unwrap();

        generate_self_signed_cert(&cert_path, &key_path)
            .expect("generation must succeed by replacing the symlink atomically");

        // The attacker target must NOT have been overwritten.
        let attacker_after = std::fs::read(&attacker_target).unwrap();
        assert_eq!(
            attacker_after, b"untouched",
            "the planted symlink must not have leaked the private key to the attacker target"
        );

        // The key path must be a regular file (not a symlink) and at 0o600.
        let key_metadata = std::fs::symlink_metadata(&key_path).unwrap();
        assert!(
            key_metadata.file_type().is_file(),
            "key path must be a regular file after generation, not a symlink"
        );
        assert_eq!(
            key_metadata.permissions().mode() & 0o777,
            0o600,
            "key file must be created at 0o600 in a single syscall, not chmod'd after"
        );

        // The key on disk must contain the freshly-generated PEM.
        let key_content = std::fs::read_to_string(&key_path).unwrap();
        assert!(key_content.contains("BEGIN PRIVATE KEY"));
    }

    #[cfg(unix)]
    #[test]
    fn test_generate_self_signed_cert_does_not_follow_planted_cert_symlink() {
        // SECURITY regression: same threat class as the key-side
        // test, but for the cert path. The cert itself is public so
        // disclosure is moot, but the redirect-write primitive
        // could be weaponized to plant attacker-chosen content at
        // an attacker-chosen path that the daemon's identity can
        // write (trust-on-first-use for any later reader of that
        // path).
        use std::os::unix::fs::symlink;

        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");
        let attacker_target = dir.path().join("attacker-target");
        std::fs::write(&attacker_target, b"untouched").unwrap();
        symlink(&attacker_target, &cert_path).unwrap();

        generate_self_signed_cert(&cert_path, &key_path)
            .expect("generation must succeed by replacing the cert symlink atomically");

        let attacker_after = std::fs::read(&attacker_target).unwrap();
        assert_eq!(
            attacker_after, b"untouched",
            "the planted cert symlink must not have leaked content to the attacker target"
        );
        let cert_metadata = std::fs::symlink_metadata(&cert_path).unwrap();
        assert!(
            cert_metadata.file_type().is_file(),
            "cert path must be a regular file after generation"
        );
    }

    #[test]
    fn test_load_generated_certs() {
        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        // Load them back
        let certs = load_certs(&cert_path).unwrap();
        assert!(!certs.is_empty());
        assert_eq!(certs.len(), 1);

        let key = load_private_key(&key_path).unwrap();
        assert!(matches!(key, PrivateKeyDer::Pkcs8(_)));
    }

    #[test]
    fn test_compute_fingerprint() {
        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        let certs = load_certs(&cert_path).unwrap();
        let fingerprint = compute_cert_fingerprint(&certs[0]);

        // Verify fingerprint format: colon-separated hex pairs, 32 bytes = 32 pairs
        let parts: Vec<&str> = fingerprint.split(':').collect();
        assert_eq!(parts.len(), 32, "SHA-256 should produce 32 bytes");

        // Each part should be exactly 2 hex characters
        for part in &parts {
            assert_eq!(part.len(), 2);
            assert!(part.chars().all(|c| c.is_ascii_hexdigit()));
        }
    }

    #[test]
    fn test_fingerprint_deterministic() {
        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        let certs = load_certs(&cert_path).unwrap();
        let fp1 = compute_cert_fingerprint(&certs[0]);
        let fp2 = compute_cert_fingerprint(&certs[0]);
        assert_eq!(fp1, fp2, "Fingerprint should be deterministic");
    }

    #[test]
    fn test_setup_tls_auto_generate() {
        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        // Pre-generate certs so setup_tls can load them
        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        let config = TlsConfig {
            enabled: true,
            cert_path: Some(cert_path.clone()),
            key_path: Some(key_path.clone()),
            auto_generate: false,
        };

        let result = setup_tls(&config).unwrap();
        assert!(!result.fingerprint.is_empty());
        assert_eq!(result.cert_path, cert_path);
        assert_eq!(result.key_path, key_path);
    }

    #[test]
    fn test_setup_tls_missing_cert_no_auto() {
        let config = TlsConfig {
            enabled: true,
            cert_path: None,
            key_path: None,
            auto_generate: false,
        };

        let result = setup_tls(&config);
        assert!(result.is_err());
    }

    #[test]
    fn test_setup_tls_partial_paths_no_auto() {
        let config = TlsConfig {
            enabled: true,
            cert_path: Some(PathBuf::from("/some/cert.pem")),
            key_path: None,
            auto_generate: false,
        };

        let result = setup_tls(&config);
        assert!(result.is_err());
    }

    #[test]
    fn test_load_certs_nonexistent() {
        let result = load_certs(Path::new("/nonexistent/cert.pem"));
        assert!(result.is_err());
    }

    #[test]
    fn test_load_key_nonexistent() {
        let result = load_private_key(Path::new("/nonexistent/key.pem"));
        assert!(result.is_err());
    }

    #[test]
    fn test_load_certs_empty_file() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("empty.pem");
        std::fs::write(&path, "").unwrap();

        let result = load_certs(&path);
        assert!(matches!(result, Err(TlsError::NoCertsFound(_))));
    }

    #[test]
    fn test_load_key_empty_file() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("empty.pem");
        std::fs::write(&path, "").unwrap();

        let result = load_private_key(&path);
        assert!(matches!(result, Err(TlsError::NoKeyFound(_))));
    }

    #[test]
    fn test_default_tls_dir() {
        let dir = default_tls_dir();
        // Should end with carapace/tls
        assert!(dir.ends_with("carapace/tls") || dir.ends_with("carapace\\tls"));
    }

    #[cfg(unix)]
    #[test]
    fn test_key_file_permissions() {
        use std::os::unix::fs::PermissionsExt;

        let dir = TempDir::new().unwrap();
        let cert_path = dir.path().join("cert.pem");
        let key_path = dir.path().join("key.pem");

        generate_self_signed_cert(&cert_path, &key_path).unwrap();

        let metadata = std::fs::metadata(&key_path).unwrap();
        let mode = metadata.permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "Key file should have 600 permissions");
    }

    // ====================================================================
    // mTLS config parsing tests
    // ====================================================================

    #[test]
    fn test_mtls_config_default() {
        let config = MtlsConfig::default();
        assert!(!config.enabled);
        assert!(config.ca_cert.is_none());
        assert!(config.crl_path.is_none());
        assert!(config.node_cert.is_none());
        assert!(config.node_key.is_none());
        assert!(config.require_client_cert);
    }

    #[test]
    fn test_parse_mtls_config_empty() {
        let cfg = serde_json::json!({});
        let mtls = parse_mtls_config(&cfg);
        assert!(!mtls.enabled);
        assert!(mtls.ca_cert.is_none());
        assert!(mtls.crl_path.is_none());
        assert!(mtls.node_cert.is_none());
        assert!(mtls.node_key.is_none());
        assert!(mtls.require_client_cert);
    }

    #[test]
    fn test_parse_mtls_config_enabled() {
        let cfg = serde_json::json!({
            "gateway": {
                "mtls": {
                    "enabled": true,
                    "caCert": "/path/to/ca.pem",
                    "crlPath": "/path/to/crl.json",
                    "nodeCert": "/path/to/node-cert.pem",
                    "nodeKey": "/path/to/node-key.pem",
                    "requireClientCert": false
                }
            }
        });
        let mtls = parse_mtls_config(&cfg);
        assert!(mtls.enabled);
        assert_eq!(mtls.ca_cert, Some(PathBuf::from("/path/to/ca.pem")));
        assert_eq!(mtls.crl_path, Some(PathBuf::from("/path/to/crl.json")));
        assert_eq!(
            mtls.node_cert,
            Some(PathBuf::from("/path/to/node-cert.pem"))
        );
        assert_eq!(mtls.node_key, Some(PathBuf::from("/path/to/node-key.pem")));
        assert!(!mtls.require_client_cert);
    }

    #[test]
    fn test_parse_mtls_config_defaults() {
        let cfg = serde_json::json!({
            "gateway": {
                "mtls": {
                    "enabled": true
                }
            }
        });
        let mtls = parse_mtls_config(&cfg);
        assert!(mtls.enabled);
        assert!(mtls.ca_cert.is_none());
        assert!(mtls.crl_path.is_none());
        assert!(mtls.node_cert.is_none());
        assert!(mtls.node_key.is_none());
        assert!(mtls.require_client_cert); // default true
    }

    #[test]
    fn test_setup_mtls_missing_ca_cert() {
        let config = MtlsConfig {
            enabled: true,
            ca_cert: None,
            crl_path: None,
            node_cert: Some(PathBuf::from("/some/cert.pem")),
            node_key: Some(PathBuf::from("/some/key.pem")),
            require_client_cert: true,
        };
        let result = setup_mtls(&config);
        assert!(result.is_err());
    }

    #[test]
    fn test_setup_mtls_missing_node_cert() {
        let config = MtlsConfig {
            enabled: true,
            ca_cert: Some(PathBuf::from("/some/ca.pem")),
            crl_path: None,
            node_cert: None,
            node_key: Some(PathBuf::from("/some/key.pem")),
            require_client_cert: true,
        };
        let result = setup_mtls(&config);
        assert!(result.is_err());
    }

    #[test]
    fn test_setup_mtls_missing_node_key() {
        let config = MtlsConfig {
            enabled: true,
            ca_cert: Some(PathBuf::from("/some/ca.pem")),
            crl_path: None,
            node_cert: Some(PathBuf::from("/some/cert.pem")),
            node_key: None,
            require_client_cert: true,
        };
        let result = setup_mtls(&config);
        assert!(result.is_err());
    }

    #[test]
    fn test_setup_mtls_with_valid_certs() {
        let dir = TempDir::new().unwrap();

        // Generate a cluster CA and node cert
        let ca = ca::ClusterCA::generate(dir.path()).unwrap();
        let node_dir = dir.path().join("nodes");
        let node_cert = ca.issue_node_cert("test-node", &node_dir).unwrap();

        let config = MtlsConfig {
            enabled: true,
            ca_cert: Some(ca.ca_cert_path()),
            crl_path: None,
            node_cert: Some(node_cert.cert_path.clone()),
            node_key: Some(node_cert.key_path.clone()),
            require_client_cert: true,
        };

        let result = setup_mtls(&config).unwrap();
        assert!(!result.node_fingerprint.is_empty());
        assert!(!result.ca_fingerprint.is_empty());
        assert_ne!(result.node_fingerprint, result.ca_fingerprint);
    }

    #[test]
    fn test_mtls_revoked_client_cert_rejected() {
        let dir = TempDir::new().unwrap();

        let ca = ca::ClusterCA::generate(dir.path()).unwrap();
        let node_dir = dir.path().join("nodes");
        let node_cert = ca.issue_node_cert("node-revoked", &node_dir).unwrap();

        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();

        let ca_certs = load_certs(&ca.ca_cert_path()).unwrap();
        let node_certs = load_certs(&node_cert.cert_path).unwrap();

        let mut root_store = rustls::RootCertStore::empty();
        root_store.add(ca_certs[0].clone()).unwrap();

        let client_verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
            .build()
            .unwrap();

        let mut revoked = HashSet::new();
        revoked.insert(compute_cert_fingerprint(&node_certs[0]).to_ascii_lowercase());
        let verifier = CrlClientCertVerifier::new(client_verifier, revoked);

        let result = verifier.verify_client_cert(&node_certs[0], &[], UnixTime::now());
        assert!(matches!(
            result,
            Err(rustls::Error::InvalidCertificate(
                rustls::CertificateError::Revoked
            ))
        ));
    }

    #[test]
    fn test_setup_mtls_optional_client_cert() {
        let dir = TempDir::new().unwrap();

        let ca = ca::ClusterCA::generate(dir.path()).unwrap();
        let node_dir = dir.path().join("nodes");
        let node_cert = ca.issue_node_cert("test-node-opt", &node_dir).unwrap();

        let config = MtlsConfig {
            enabled: true,
            ca_cert: Some(ca.ca_cert_path()),
            crl_path: None,
            node_cert: Some(node_cert.cert_path.clone()),
            node_key: Some(node_cert.key_path.clone()),
            require_client_cert: false,
        };

        let result = setup_mtls(&config);
        assert!(result.is_ok());
    }
}
