nexar 0.1.2

use crate::error::{NexarError, Result};
use std::sync::Arc;

/// Ephemeral cluster CA for mutual TLS.
///
/// Generated by the seed node at cluster formation time. Signs per-node
/// leaf certificates so all mesh connections can verify each other.
/// The CA key never leaves the seed process.
pub struct ClusterCa {
    cert_der: rustls::pki_types::CertificateDer<'static>,
    /// Issuer params recovered from the generate path. `None` after a
    /// `from_der` reload — in that case we rebuild the issuer at
    /// issue-time via `rcgen::Issuer::from_ca_cert_der`, which parses
    /// the DN + key-usage bits from the stored cert DER.
    params: Option<rcgen::CertificateParams>,
    key_pair: rcgen::KeyPair,
}

impl ClusterCa {
    /// Generate a new ephemeral CA keypair.
    pub fn generate() -> Result<Self> {
        let mut params = rcgen::CertificateParams::new(Vec::<String>::new())
            .map_err(|e| NexarError::Tls(e.to_string()))?;
        params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        params.distinguished_name = rcgen::DistinguishedName::new();
        params
            .distinguished_name
            .push(rcgen::DnType::CommonName, "nexar-cluster-ca");

        let key_pair = rcgen::KeyPair::generate().map_err(|e| NexarError::Tls(e.to_string()))?;
        let cert = params
            .self_signed(&key_pair)
            .map_err(|e| NexarError::Tls(e.to_string()))?;

        let cert_der = rustls::pki_types::CertificateDer::from(cert.der().to_vec());
        Ok(Self {
            cert_der,
            params: Some(params),
            key_pair,
        })
    }

    /// DER-encoded CA certificate (shared with all nodes as trust anchor).
    pub fn cert_der(&self) -> rustls::pki_types::CertificateDer<'static> {
        self.cert_der.clone()
    }

    /// Issue a leaf certificate signed by this CA with a single SAN.
    ///
    /// Equivalent to [`issue_cert_multi`] with `&[san]`; kept as a
    /// convenience for single-SAN callers.
    pub fn issue_cert(
        &self,
        san: &str,
    ) -> Result<(
        rustls::pki_types::CertificateDer<'static>,
        rustls::pki_types::PrivateKeyDer<'static>,
    )> {
        self.issue_cert_multi(&[san])
    }

    /// Issue a leaf certificate signed by this CA with one or more
    /// Subject Alternative Names.
    ///
    /// Every string in `sans` is added to the leaf's SAN list so the
    /// same cert can satisfy multiple SNI hostnames (e.g. the fixed
    /// cluster SNI `"nodedb"` *and* the per-node identity
    /// `"node-3"`). Returns `(cert_der, key_der)` for the node.
    pub fn issue_cert_multi(
        &self,
        sans: &[&str],
    ) -> Result<(
        rustls::pki_types::CertificateDer<'static>,
        rustls::pki_types::PrivateKeyDer<'static>,
    )> {
        if sans.is_empty() {
            return Err(NexarError::Tls(
                "issue_cert_multi requires at least one SAN".into(),
            ));
        }
        let san_vec: Vec<String> = sans.iter().map(|s| (*s).to_string()).collect();
        let leaf_params =
            rcgen::CertificateParams::new(san_vec).map_err(|e| NexarError::Tls(e.to_string()))?;
        let leaf_key = rcgen::KeyPair::generate().map_err(|e| NexarError::Tls(e.to_string()))?;

        // On the generate path we still hold the original
        // `CertificateParams`, which carries the exact DN + key
        // usage bits — build the issuer from those. After a
        // `from_der` reload `params` is None and we rebuild the
        // issuer from the stored cert DER via x509-parser.
        let leaf_cert = match &self.params {
            Some(params) => {
                let issuer = rcgen::Issuer::from_params(params, &self.key_pair);
                leaf_params
                    .signed_by(&leaf_key, &issuer)
                    .map_err(|e| NexarError::Tls(e.to_string()))?
            }
            None => {
                let issuer = rcgen::Issuer::from_ca_cert_der(&self.cert_der, &self.key_pair)
                    .map_err(|e| NexarError::Tls(format!("rebuild issuer: {e}")))?;
                leaf_params
                    .signed_by(&leaf_key, &issuer)
                    .map_err(|e| NexarError::Tls(e.to_string()))?
            }
        };

        let cert_der = rustls::pki_types::CertificateDer::from(leaf_cert.der().to_vec());
        let key_der = rustls::pki_types::PrivateKeyDer::try_from(leaf_key.serialize_der())
            .map_err(|e| NexarError::Tls(e.to_string()))?;

        Ok((cert_der, key_der))
    }

    /// PKCS#8 DER-encoded private key of this CA. Exposes the key so
    /// the caller can persist it (e.g. on disk at 0600) and later
    /// reconstruct the `ClusterCa` via [`from_der`] to issue further
    /// leaf certificates under the same CA — used by operator tooling
    /// that rotates node certs without rotating the CA itself.
    ///
    /// Treat the return value as key material: never log, zeroize on
    /// drop where the caller's threat model requires it.
    pub fn key_pair_pkcs8_der(&self) -> Vec<u8> {
        self.key_pair.serialize_der()
    }

    /// Reconstruct a `ClusterCa` from a previously-saved
    /// `(key_pair_pkcs8_der, cert_der)` pair. Inverse of
    /// [`key_pair_pkcs8_der`] + [`cert_der`].
    ///
    /// Requires the `x509-parser` feature on rcgen (enabled by this
    /// crate's default build).
    pub fn from_der(
        key_pair_pkcs8_der: &[u8],
        cert_der: &rustls::pki_types::CertificateDer<'_>,
    ) -> Result<Self> {
        let key_pair = rcgen::KeyPair::try_from(key_pair_pkcs8_der)
            .map_err(|e| NexarError::Tls(format!("load CA key: {e}")))?;
        Ok(Self {
            cert_der: cert_der.clone().into_owned(),
            params: None,
            key_pair,
        })
    }
}

/// Create a quinn `TransportConfig` optimized for datacenter networks.
///
/// Datacenter networks have low latency, high bandwidth, and minimal random
/// packet loss — different assumptions from internet QUIC defaults.
pub(crate) fn datacenter_transport_config() -> quinn::TransportConfig {
    let mut config = quinn::TransportConfig::default();
    // 256 MiB connection-level receive window (vs default ~1.5 MiB).
    config.receive_window(quinn::VarInt::from_u32(256 * 1024 * 1024));
    // 256 MiB send window.
    config.send_window(256 * 1024 * 1024);
    // 64 MiB per-stream receive window.
    config.stream_receive_window(quinn::VarInt::from_u32(64 * 1024 * 1024));
    // Low initial RTT estimate for datacenter (100 µs vs default 333 ms).
    config.initial_rtt(std::time::Duration::from_micros(100));
    // Aggressive keep-alive (datacenter links are reliable).
    config.keep_alive_interval(Some(std::time::Duration::from_secs(5)));
    // Short idle timeout.
    config.max_idle_timeout(Some(
        std::time::Duration::from_secs(30)
            .try_into()
            .expect("30s fits in IdleTimeout"),
    ));
    // Allow many concurrent unidirectional streams. Each tagged send opens a
    // new uni stream, and the stream pool pre-opens 8 per connection.
    // Default (quinn 0.11) is 0 for uni streams, which blocks all sends.
    config.max_concurrent_uni_streams(quinn::VarInt::from_u32(1024));
    config
}

/// Generate a self-signed certificate for the seed's bootstrap listener.
///
/// Used only during cluster formation (before CA exists). Workers connect
/// to this with [`make_bootstrap_client_config`] which skips verification.
pub fn generate_self_signed_cert() -> Result<(
    rustls::pki_types::CertificateDer<'static>,
    rustls::pki_types::PrivateKeyDer<'static>,
)> {
    let cert_params = rcgen::CertificateParams::new(vec!["localhost".into()])
        .map_err(|e| NexarError::Tls(e.to_string()))?;
    let key_pair = rcgen::KeyPair::generate().map_err(|e| NexarError::Tls(e.to_string()))?;
    let cert = cert_params
        .self_signed(&key_pair)
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    let cert_der = rustls::pki_types::CertificateDer::from(cert.der().to_vec());
    let key_der = rustls::pki_types::PrivateKeyDer::try_from(key_pair.serialize_der())
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    Ok((cert_der, key_der))
}

/// Build a `ServerConfig` for the seed's bootstrap listener (no client auth).
///
/// Used only during cluster formation. All mesh connections use
/// [`make_server_config_mtls`] instead.
pub fn make_server_config(
    cert: rustls::pki_types::CertificateDer<'static>,
    key: rustls::pki_types::PrivateKeyDer<'static>,
) -> Result<quinn::ServerConfig> {
    let mut tls_config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    tls_config.alpn_protocols = vec![b"nexar/1".to_vec()];

    let quic_config = quinn::crypto::rustls::QuicServerConfig::try_from(Arc::new(tls_config))
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    let mut server_config = quinn::ServerConfig::with_crypto(Arc::new(quic_config));
    server_config.transport_config(Arc::new(datacenter_transport_config()));
    Ok(server_config)
}

/// Build a `ServerConfig` with mutual TLS — requires client certs signed by the cluster CA.
///
/// Used for all mesh (P2P) listeners after cluster formation.
pub fn make_server_config_mtls(
    cert: rustls::pki_types::CertificateDer<'static>,
    key: rustls::pki_types::PrivateKeyDer<'static>,
    ca_cert: &rustls::pki_types::CertificateDer<'static>,
) -> Result<quinn::ServerConfig> {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(ca_cert.clone())
        .map_err(|e| NexarError::Tls(format!("add CA to root store: {e}")))?;

    let client_verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
        .build()
        .map_err(|e| NexarError::Tls(format!("build client verifier: {e}")))?;

    let mut tls_config = rustls::ServerConfig::builder()
        .with_client_cert_verifier(client_verifier)
        .with_single_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    tls_config.alpn_protocols = vec![b"nexar/1".to_vec()];

    let quic_config = quinn::crypto::rustls::QuicServerConfig::try_from(Arc::new(tls_config))
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    let mut server_config = quinn::ServerConfig::with_crypto(Arc::new(quic_config));
    server_config.transport_config(Arc::new(datacenter_transport_config()));
    Ok(server_config)
}

/// Build a `ClientConfig` that skips server verification (bootstrap only).
///
/// Used only for the initial worker→seed connection during cluster formation.
/// All mesh connections use [`make_client_config_mtls`] instead.
pub fn make_bootstrap_client_config() -> Result<quinn::ClientConfig> {
    let mut tls_config = rustls::ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(SkipServerVerification))
        .with_no_client_auth();

    tls_config.alpn_protocols = vec![b"nexar/1".to_vec()];

    let quic_config = quinn::crypto::rustls::QuicClientConfig::try_from(Arc::new(tls_config))
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    let mut client_config = quinn::ClientConfig::new(Arc::new(quic_config));
    client_config.transport_config(Arc::new(datacenter_transport_config()));
    Ok(client_config)
}

/// Build a `ClientConfig` with mutual TLS — verifies server cert and presents client cert.
///
/// Used for all mesh (P2P) connections after cluster formation.
pub fn make_client_config_mtls(
    cert: rustls::pki_types::CertificateDer<'static>,
    key: rustls::pki_types::PrivateKeyDer<'static>,
    ca_cert: &rustls::pki_types::CertificateDer<'static>,
) -> Result<quinn::ClientConfig> {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(ca_cert.clone())
        .map_err(|e| NexarError::Tls(format!("add CA to root store: {e}")))?;

    let mut tls_config = rustls::ClientConfig::builder()
        .with_root_certificates(root_store)
        .with_client_auth_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    tls_config.alpn_protocols = vec![b"nexar/1".to_vec()];

    let quic_config = quinn::crypto::rustls::QuicClientConfig::try_from(Arc::new(tls_config))
        .map_err(|e| NexarError::Tls(e.to_string()))?;

    let mut client_config = quinn::ClientConfig::new(Arc::new(quic_config));
    client_config.transport_config(Arc::new(datacenter_transport_config()));
    Ok(client_config)
}

/// Certificate verifier that accepts any certificate.
///
/// **Bootstrap only** — used for the initial worker→seed connection before
/// the worker has received CA-signed credentials. All subsequent mesh
/// connections use proper CA-based verification.
///
/// # MITM mitigation
///
/// Since this verifier skips certificate validation, the bootstrap channel
/// is protected by a pre-shared cluster token (`NEXAR_CLUSTER_TOKEN` env var).
/// When set, both seed and workers must present the same token in the Hello
/// message. An attacker intercepting the connection cannot forge the token.
#[derive(Debug)]
struct SkipServerVerification;

impl rustls::client::danger::ServerCertVerifier for SkipServerVerification {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> std::result::Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
        Ok(rustls::client::danger::ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &rustls::DigitallySignedStruct,
    ) -> std::result::Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &rustls::DigitallySignedStruct,
    ) -> std::result::Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
        rustls::crypto::ring::default_provider()
            .signature_verification_algorithms
            .supported_schemes()
    }
}

/// Build a raw `rustls::ServerConfig` for the TCP bulk sidecar with mutual TLS.
///
/// Unlike [`make_server_config_mtls`] which wraps into a QUIC config, this
/// returns a plain rustls config suitable for wrapping TCP streams via
/// `tokio-rustls`. Uses ALPN `nexar-bulk/1` to distinguish from QUIC traffic.
pub fn make_bulk_tls_server_config(
    cert: rustls::pki_types::CertificateDer<'static>,
    key: rustls::pki_types::PrivateKeyDer<'static>,
    ca_cert: &rustls::pki_types::CertificateDer<'static>,
) -> Result<Arc<rustls::ServerConfig>> {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(ca_cert.clone())
        .map_err(|e| NexarError::Tls(format!("bulk TLS: add CA to root store: {e}")))?;

    let client_verifier = rustls::server::WebPkiClientVerifier::builder(Arc::new(root_store))
        .build()
        .map_err(|e| NexarError::Tls(format!("bulk TLS: build client verifier: {e}")))?;

    let mut tls_config = rustls::ServerConfig::builder()
        .with_client_cert_verifier(client_verifier)
        .with_single_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(format!("bulk TLS server config: {e}")))?;

    tls_config.alpn_protocols = vec![b"nexar-bulk/1".to_vec()];
    Ok(Arc::new(tls_config))
}

/// Build a raw `rustls::ClientConfig` for the TCP bulk sidecar with mutual TLS.
///
/// Returns a plain rustls config suitable for wrapping TCP streams via
/// `tokio-rustls`. Uses ALPN `nexar-bulk/1`.
pub fn make_bulk_tls_client_config(
    cert: rustls::pki_types::CertificateDer<'static>,
    key: rustls::pki_types::PrivateKeyDer<'static>,
    ca_cert: &rustls::pki_types::CertificateDer<'static>,
) -> Result<Arc<rustls::ClientConfig>> {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(ca_cert.clone())
        .map_err(|e| NexarError::Tls(format!("bulk TLS: add CA to root store: {e}")))?;

    let mut tls_config = rustls::ClientConfig::builder()
        .with_root_certificates(root_store)
        .with_client_auth_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(format!("bulk TLS client config: {e}")))?;

    tls_config.alpn_protocols = vec![b"nexar-bulk/1".to_vec()];
    Ok(Arc::new(tls_config))
}

/// Build a `rustls::ServerConfig` that skips client auth for the TCP bulk sidecar.
///
/// Used in `bootstrap_local` where all nodes share self-signed certs.
/// Provides encryption without authentication (same trust boundary as QUIC bootstrap).
pub(crate) fn make_bulk_tls_server_config_insecure() -> Result<Arc<rustls::ServerConfig>> {
    let (cert, key) = generate_self_signed_cert()?;

    let mut tls_config = rustls::ServerConfig::builder()
        .with_no_client_auth()
        .with_single_cert(vec![cert], key)
        .map_err(|e| NexarError::Tls(format!("bulk TLS insecure server: {e}")))?;

    tls_config.alpn_protocols = vec![b"nexar-bulk/1".to_vec()];
    Ok(Arc::new(tls_config))
}

/// Build a `rustls::ClientConfig` that skips server verification for the TCP bulk sidecar.
///
/// Used in `bootstrap_local` where all nodes share self-signed certs.
pub(crate) fn make_bulk_tls_client_config_insecure() -> Result<Arc<rustls::ClientConfig>> {
    let mut tls_config = rustls::ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(SkipServerVerification))
        .with_no_client_auth();

    tls_config.alpn_protocols = vec![b"nexar-bulk/1".to_vec()];
    Ok(Arc::new(tls_config))
}

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

    #[test]
    fn test_generate_self_signed_cert() {
        let (cert, key) = generate_self_signed_cert().unwrap();
        assert!(!cert.is_empty());
        let _ = key;
    }

    #[test]
    fn test_make_server_config() {
        let (cert, key) = generate_self_signed_cert().unwrap();
        let config = make_server_config(cert, key).unwrap();
        let _ = config;
    }

    #[test]
    fn test_make_bootstrap_client_config() {
        let config = make_bootstrap_client_config().unwrap();
        let _ = config;
    }

    #[test]
    fn test_ca_generation() {
        let ca = ClusterCa::generate().unwrap();
        let cert_der = ca.cert_der();
        assert!(!cert_der.is_empty());
    }

    #[test]
    fn test_issue_cert_signed_by_ca() {
        let ca = ClusterCa::generate().unwrap();
        let (cert, key) = ca.issue_cert("localhost").unwrap();
        assert!(!cert.is_empty());
        let _ = key;
    }

    #[test]
    fn test_mtls_server_config_with_ca() {
        let ca = ClusterCa::generate().unwrap();
        let ca_cert = ca.cert_der();
        let (cert, key) = ca.issue_cert("localhost").unwrap();
        let config = make_server_config_mtls(cert, key, &ca_cert).unwrap();
        let _ = config;
    }

    #[test]
    fn test_mtls_client_config_with_ca() {
        let ca = ClusterCa::generate().unwrap();
        let ca_cert = ca.cert_der();
        let (cert, key) = ca.issue_cert("localhost").unwrap();
        let config = make_client_config_mtls(cert, key, &ca_cert).unwrap();
        let _ = config;
    }

    #[test]
    fn test_multiple_leaf_certs_from_same_ca() {
        let ca = ClusterCa::generate().unwrap();
        let (cert1, _key1) = ca.issue_cert("localhost").unwrap();
        let (cert2, _key2) = ca.issue_cert("localhost").unwrap();
        // Each call generates a unique cert (different key pair).
        assert_ne!(cert1.as_ref(), cert2.as_ref());
    }

    #[test]
    fn test_issue_cert_multi_rejects_empty() {
        let ca = ClusterCa::generate().unwrap();
        assert!(ca.issue_cert_multi(&[]).is_err());
    }

    #[test]
    fn test_issue_cert_multi_with_multiple_sans() {
        let ca = ClusterCa::generate().unwrap();
        let (cert, _key) = ca.issue_cert_multi(&["nodedb", "node-3"]).unwrap();
        assert!(!cert.is_empty());
    }

    #[test]
    fn test_ca_roundtrip_via_der() {
        let ca1 = ClusterCa::generate().unwrap();
        let ca1_cert = ca1.cert_der();
        let ca1_key = ca1.key_pair_pkcs8_der();
        assert!(!ca1_key.is_empty());

        // Reconstruct and verify the reloaded CA still issues certs
        // that chain to the original anchor.
        let ca2 = ClusterCa::from_der(&ca1_key, &ca1_cert).unwrap();
        let (leaf, _leaf_key) = ca2.issue_cert("nodedb").unwrap();

        // mTLS server config built from the ORIGINAL CA cert accepts
        // a leaf issued by the RELOADED CA — proof that the key
        // pair survived the round trip and the two CA instances
        // are the same issuer.
        let config = make_server_config_mtls(leaf, _leaf_key, &ca1_cert).unwrap();
        let _ = config;
    }
}