rmcp_server_kit/

mtls_revocation.rs

//! CDP-driven CRL revocation support for mTLS.
//!
//! When mTLS is configured with CRL checks enabled, startup performs a bounded
//! bootstrap pass over the configured CA bundle, extracts CRL Distribution
//! Point (CDP) URLs, fetches reachable CRLs, and builds the initial inner
//! `rustls` verifier from that cache.
//!
//! During handshakes, the outer verifier remains stable for the lifetime of the
//! TLS acceptor while its inner `WebPkiClientVerifier` is swapped atomically via
//! `ArcSwap` as CRLs are discovered or refreshed. Discovery from connecting
//! client certificates is fire-and-forget and never blocks the synchronous
//! handshake path.
//!
//! Security note: CDP URLs are extracted from attacker-controllable client
//! certs *before* chain validation. This is safe by design; see the
//! `// SECURITY:` comment on `DynamicClientCertVerifier::verify_client_cert`
//! for the full rationale before changing the discovery ordering.
//!
//! Semantics:
//! - `crl_deny_on_unavailable = false` => fail open with warn logs.
//! - `crl_deny_on_unavailable = true` => fail closed when a certificate
//!   advertises CDP URLs whose revocation status is not yet available.

use std::{
    collections::{HashMap, HashSet},
    num::NonZeroU32,
    pin::Pin,
    sync::{Arc, Mutex},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use arc_swap::ArcSwap;
use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
use rustls::{
    DigitallySignedStruct, DistinguishedName, Error as TlsError, RootCertStore, SignatureScheme,
    client::danger::HandshakeSignatureValid,
    pki_types::{CertificateDer, CertificateRevocationListDer, UnixTime},
    server::{
        WebPkiClientVerifier,
        danger::{ClientCertVerified, ClientCertVerifier},
    },
};
use tokio::{
    net::lookup_host,
    sync::{RwLock, Semaphore, mpsc},
    task::JoinSet,
    time::{Instant, Sleep},
};
use tokio_util::sync::CancellationToken;
use url::Url;
use x509_parser::{
    extensions::{DistributionPointName, GeneralName, ParsedExtension},
    prelude::{FromDer, X509Certificate},
    revocation_list::CertificateRevocationList,
};

use crate::{
    auth::MtlsConfig,
    error::RmcpServerKitError,
    ssrf::{check_scheme, ip_block_reason, sanitized_url_for_log},
};

const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(10);
const MIN_AUTO_REFRESH: Duration = Duration::from_mins(10);
const MAX_AUTO_REFRESH: Duration = Duration::from_hours(24);
/// Connection timeout for CRL HTTP fetches. Independent of overall fetch
/// timeout to bound time spent on unreachable hosts.
const CRL_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);

/// Parsed CRL cached in memory and keyed by its source URL.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct CachedCrl {
    /// DER bytes for the CRL.
    pub der: CertificateRevocationListDer<'static>,
    /// `thisUpdate` field from the CRL.
    pub this_update: SystemTime,
    /// `nextUpdate` field from the CRL, if present.
    pub next_update: Option<SystemTime>,
    /// Time the server fetched this CRL.
    pub fetched_at: SystemTime,
    /// Source URL used for retrieval.
    pub source_url: String,
}

pub(crate) struct VerifierHandle(pub Arc<dyn ClientCertVerifier>);

impl std::fmt::Debug for VerifierHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("VerifierHandle").finish_non_exhaustive()
    }
}

/// Shared CRL state backing the dynamic mTLS verifier.
#[allow(
    missing_debug_implementations,
    reason = "contains ArcSwap and dyn verifier internals"
)]
#[non_exhaustive]
pub struct CrlSet {
    inner_verifier: ArcSwap<VerifierHandle>,
    /// Cached CRLs keyed by URL.
    pub cache: RwLock<HashMap<String, CachedCrl>>,
    /// Immutable client-auth root store.
    pub roots: Arc<RootCertStore>,
    /// mTLS CRL configuration.
    pub config: MtlsConfig,
    /// Fire-and-forget discovery channel for newly-seen CDP URLs.
    pub discover_tx: mpsc::UnboundedSender<String>,
    client: reqwest::Client,
    /// URLs whose CRL is confirmed present in `cache`. Permanent dedup: a URL
    /// here is never re-enqueued for discovery.
    seen_urls: Mutex<HashSet<String>>,
    /// URLs admitted to the discovery channel but not yet confirmed cached.
    ///
    /// This exists so a queued URL is not re-enqueued while its fetch is in
    /// flight, WITHOUT permanently suppressing it. Promotion to `seen_urls`
    /// happens only once the CRL is actually in the cache; a fetch error or a
    /// cache-cap rejection clears the entry so a later handshake can retry.
    /// Merging the two states is exactly the bug this separation fixes: a
    /// first-fetch failure would otherwise suppress the URL for the process
    /// lifetime, silently disabling revocation for that CDP.
    pending_urls: Mutex<HashSet<String>>,
    cached_urls: Mutex<HashSet<String>>,
    /// Global cap on simultaneous CRL HTTP fetches (SSRF amplification guard).
    global_fetch_sem: Arc<Semaphore>,
    /// Per-host serializer (one in-flight fetch per origin host). Bounded
    /// by `crl_max_host_semaphores`; at the cap, idle entries are evicted
    /// on demand (see [`acquire_host_semaphore`]), so the cap only rejects
    /// genuinely concurrent fetch floods and is never a permanent lockout.
    host_semaphores: Arc<tokio::sync::Mutex<HashMap<String, Arc<Semaphore>>>>,
    /// Global rate-limiter on discovery URL submissions; protects against
    /// cert-driven URL flooding by a malicious mTLS peer.
    ///
    /// Note: this ships as a process-global limiter; per-source-IP scoping
    /// is deferred to a future release because the rustls
    /// `verify_client_cert` callback does not carry a `SocketAddr` for the
    /// peer. This is a CRL-discovery limiter in the TLS verifier path —
    /// distinct from the bearer pre-auth limiter (`AuthState`), which is
    /// already keyed per-IP via a bounded keyed governor and lives in the
    /// ordinary request middleware path.
    discovery_limiter: Arc<DefaultDirectRateLimiter>,
    /// Cached cap on per-fetch response body size; copied from `config` so the
    /// hot path doesn't re-read the (rarely changing) config struct.
    max_response_bytes: u64,
    last_cap_warn: Mutex<HashMap<&'static str, Instant>>,
}

impl CrlSet {
    fn new(
        roots: Arc<RootCertStore>,
        config: MtlsConfig,
        discover_tx: mpsc::UnboundedSender<String>,
        initial_cache: HashMap<String, CachedCrl>,
    ) -> Result<Arc<Self>, RmcpServerKitError> {
        // M-H2: install the SSRF screening resolver on the CRL fetcher.
        // CRL CDP URLs come from attacker-controllable client certs and
        // their hosts are re-resolved per fetch -- exactly the TOCTOU
        // class M-H2 closes. The allowlist is empty (default-strict),
        // matching the existing CRL pre-flight posture; operators who
        // need internal CDPs would extend this with the same
        // CompiledSsrfAllowlist plumbing used by oauth.
        let allowlist = Arc::new(crate::ssrf::CompiledSsrfAllowlist::default());
        let resolver: Arc<dyn reqwest::dns::Resolve> =
            Arc::new(crate::ssrf_resolver::SsrfScreeningResolver::new(
                Arc::clone(&allowlist),
                #[cfg(any(test, feature = "test-helpers"))]
                Arc::new(std::sync::atomic::AtomicBool::new(false)),
                #[cfg(not(any(test, feature = "test-helpers")))]
                (),
            ));

        let client = reqwest::Client::builder()
            // M-H2/N1: see oauth.rs::OauthHttpClient::build for rationale.
            .no_proxy()
            .dns_resolver(Arc::clone(&resolver))
            .timeout(config.crl_fetch_timeout)
            .connect_timeout(CRL_CONNECT_TIMEOUT)
            .tcp_keepalive(None)
            .redirect(reqwest::redirect::Policy::none())
            .user_agent(format!("rmcp-server-kit/{}", env!("CARGO_PKG_VERSION")))
            .build()
            .map_err(|error| {
                RmcpServerKitError::Startup(format!("CRL HTTP client init: {error}"))
            })?;

        let initial_verifier = rebuild_verifier(&roots, &config, &initial_cache)?;
        let seen_urls = initial_cache.keys().cloned().collect::<HashSet<_>>();
        let cached_urls = seen_urls.clone();

        let concurrency = config.crl_max_concurrent_fetches.max(1);
        let global_fetch_sem = Arc::new(Semaphore::new(concurrency));
        let host_semaphores = Arc::new(tokio::sync::Mutex::new(HashMap::new()));

        let rate =
            NonZeroU32::new(config.crl_discovery_rate_per_min.max(1)).unwrap_or(NonZeroU32::MIN);
        let discovery_limiter = Arc::new(RateLimiter::direct(Quota::per_minute(rate)));

        let max_response_bytes = config.crl_max_response_bytes;

        Ok(Arc::new(Self {
            inner_verifier: ArcSwap::from_pointee(VerifierHandle(initial_verifier)),
            cache: RwLock::new(initial_cache),
            roots,
            config,
            discover_tx,
            client,
            seen_urls: Mutex::new(seen_urls),
            pending_urls: Mutex::new(HashSet::new()),
            cached_urls: Mutex::new(cached_urls),
            global_fetch_sem,
            host_semaphores,
            discovery_limiter,
            max_response_bytes,
            last_cap_warn: Mutex::new(HashMap::new()),
        }))
    }

    fn warn_cap_exceeded_throttled(&self, which: &'static str) {
        let now = Instant::now();
        let cooldown = Duration::from_mins(1);
        let should_warn = match self.last_cap_warn.lock() {
            Ok(mut guard) => {
                let should_emit = guard
                    .get(which)
                    .is_none_or(|last| now.saturating_duration_since(*last) >= cooldown);
                if should_emit {
                    guard.insert(which, now);
                }
                should_emit
            }
            Err(poisoned) => {
                let mut guard = poisoned.into_inner();
                let should_emit = guard
                    .get(which)
                    .is_none_or(|last| now.saturating_duration_since(*last) >= cooldown);
                if should_emit {
                    guard.insert(which, now);
                }
                should_emit
            }
        };

        if should_warn {
            tracing::warn!(which = which, "CRL map cap exceeded; dropping newest entry");
        }
    }

    async fn commit_cache_update_atomically(
        &self,
        inserts: Vec<(String, CachedCrl)>,
        removals: &[String],
    ) -> Result<bool, RmcpServerKitError> {
        let mut cache = self.cache.write().await;
        let mut candidate = cache.clone();
        let mut admitted_urls = Vec::new();

        // POLICY: at cap the NEWEST entry is rejected, never an existing
        // one (no LRU). Under adversarial unique-CDP churn an LRU would
        // let an attacker evict the legitimate warm set by spamming
        // throwaway CDP URLs; rejecting newcomers instead preserves
        // revocation coverage for the established CA estate. Confirmed
        // by Oracle review of the 1.13.0 rust-review fix plan.
        for (url, cached) in inserts {
            if candidate.len() >= self.config.crl_max_cache_entries && !candidate.contains_key(&url)
            {
                self.warn_cap_exceeded_throttled("cache");
                continue;
            }
            candidate.insert(url.clone(), cached);
            admitted_urls.push(url);
        }

        for url in removals {
            candidate.remove(url);
        }

        // SECURITY: `cached_urls` is the synchronous fail-closed precheck's
        // trust hint. It must never get ahead of `inner_verifier`; otherwise a
        // handshake could skip the unavailable-CRL fast-fail for a URL the live
        // rustls verifier cannot enforce. Build from the full candidate cache
        // first, then swap verifier, then publish cache/cached_urls together.
        let verifier = rebuild_verifier(&self.roots, &self.config, &candidate)?;
        self.inner_verifier
            .store(Arc::new(VerifierHandle(verifier)));
        let changed = !admitted_urls.is_empty() || !removals.is_empty();
        *cache = candidate;
        drop(cache);

        match self.cached_urls.lock() {
            Ok(mut cached_urls) => {
                for url in admitted_urls {
                    cached_urls.insert(url);
                }
                for url in removals {
                    cached_urls.remove(url);
                }
            }
            Err(poisoned) => {
                let mut cached_urls = poisoned.into_inner();
                for url in admitted_urls {
                    cached_urls.insert(url);
                }
                for url in removals {
                    cached_urls.remove(url);
                }
            }
        }

        // A removed CRL must become fully re-discoverable, so clear the URL
        // from BOTH dedup states. Clearing only `seen_urls` would leave a
        // stale `pending_urls` entry suppressing re-enqueue forever.
        {
            let mut seen = self
                .seen_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            for url in removals {
                seen.remove(url);
            }
        }
        {
            let mut pending = self
                .pending_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            for url in removals {
                pending.remove(url);
            }
        }

        Ok(changed)
    }

    /// Force an immediate refresh of all currently known CRL URLs.
    ///
    /// # Errors
    ///
    /// Returns an error if rebuilding the inner verifier fails.
    pub async fn force_refresh(&self) -> Result<(), RmcpServerKitError> {
        let urls = {
            let cache = self.cache.read().await;
            cache.keys().cloned().collect::<Vec<_>>()
        };
        self.refresh_urls(urls).await
    }

    async fn refresh_due_urls(&self) -> Result<(), RmcpServerKitError> {
        let now = SystemTime::now();
        let urls = {
            let cache = self.cache.read().await;
            cache
                .iter()
                .filter(|(_, cached)| {
                    should_refresh_cached(cached, now, self.config.crl_refresh_interval)
                })
                .map(|(url, _)| url.clone())
                .collect::<Vec<_>>()
        };

        if urls.is_empty() {
            return Ok(());
        }

        self.refresh_urls(urls).await
    }

    async fn refresh_urls(&self, urls: Vec<String>) -> Result<(), RmcpServerKitError> {
        let results = self.fetch_url_results(urls).await;
        let now = SystemTime::now();
        let cache = self.cache.read().await;
        let mut inserts = Vec::new();
        let mut removals = Vec::new();

        for (url, result) in results {
            match result {
                Ok(cached) => {
                    inserts.push((url, cached));
                }
                Err(error) => {
                    let remove_entry = cache.get(&url).is_some_and(|existing| {
                        existing
                            .next_update
                            .and_then(|next| next.checked_add(self.config.crl_stale_grace))
                            .is_some_and(|deadline| now > deadline)
                    });
                    tracing::warn!(url = %url, error = %error, "CRL refresh failed");
                    if remove_entry {
                        removals.push(url);
                    }
                }
            }
        }
        drop(cache);

        if !inserts.is_empty() || !removals.is_empty() {
            let _ = self
                .commit_cache_update_atomically(inserts, &removals)
                .await?;
        }

        Ok(())
    }

    /// Fetch a CRL and commit it to the cache.
    ///
    /// Returns whether the CRL is actually present in the cache afterwards.
    /// A successful HTTP fetch is NOT sufficient:
    /// [`Self::commit_cache_update_atomically`] rejects new entries once
    /// `crl_max_cache_entries` is reached. Only a URL that genuinely landed in
    /// the cache may be promoted to the permanent `seen_urls` dedup set —
    /// promoting on fetch success alone would suppress a URL that was never
    /// cached, which is the same revocation-bypass this state split fixes.
    async fn fetch_and_store_url(&self, url: String) -> Result<bool, RmcpServerKitError> {
        let cached = gated_fetch(
            &self.client,
            &self.global_fetch_sem,
            &self.host_semaphores,
            &url,
            self.config.crl_allow_http,
            self.max_response_bytes,
            self.config.crl_max_host_semaphores,
        )
        .await?;
        let _ = self
            .commit_cache_update_atomically(vec![(url.clone(), cached)], &[])
            .await?;
        Ok(self.cache.read().await.contains_key(&url))
    }

    /// Promote a URL from the in-flight set to the permanent dedup set.
    /// Called only once its CRL is confirmed present in the cache.
    fn promote_pending_to_seen(&self, url: &str) {
        {
            let mut pending = self
                .pending_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            pending.remove(url);
        }
        let mut seen = self
            .seen_urls
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if seen.len() >= self.config.crl_max_seen_urls && !seen.contains(url) {
            self.warn_cap_exceeded_throttled("seen_urls");
            return;
        }
        seen.insert(url.to_owned());
    }

    /// Clear a URL's in-flight marker without promoting it, so a later
    /// handshake can re-enqueue it. Used when the fetch failed or the cache
    /// refused the entry.
    fn clear_pending(&self, url: &str) {
        let mut pending = self
            .pending_urls
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        pending.remove(url);
    }

    fn note_discovered_urls(
        &self,
        end_entity_urls: &[String],
        intermediate_urls: &[String],
    ) -> bool {
        // INVARIANT: only called post-handshake from
        // `DynamicClientCertVerifier::verify_client_cert`. The peer has
        // already presented a chain that parses; this method must not panic
        // under attacker-controlled URL contents.
        //
        // SECURITY: see `DynamicClientCertVerifier::verify_client_cert` for
        // the rationale on why accepting URLs from an unverified cert is
        // safe (no HTTP on this path; fetch is off-path and SSRF-gated).
        let mut all_urls = Vec::with_capacity(end_entity_urls.len() + intermediate_urls.len());
        all_urls.extend_from_slice(end_entity_urls);
        all_urls.extend_from_slice(intermediate_urls);
        all_urls.sort();
        all_urls.dedup();

        // Snapshot both dedup sets under their locks; do NOT mutate yet.
        // A URL is skipped if it is already cached (`seen_urls`) or already
        // queued and awaiting its fetch (`pending_urls`). Promotion to
        // `seen_urls` happens only after the CRL is confirmed in the cache,
        // so a URL that loses the limiter race, hits a closed channel, fails
        // to fetch, or is rejected by the cache cap stays retriable. Marking
        // "seen" any earlier permanently black-holes the URL: every later
        // handshake would treat it as known and skip discovery, while no CRL
        // was ever cached. With `crl_deny_on_unavailable = true` that is a
        // persistent handshake failure; with fail-open it silently disables
        // revocation checking for that CDP for the process lifetime.
        let candidates: Vec<String> = {
            let seen = self
                .seen_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let pending = self
                .pending_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            all_urls
                .iter()
                .filter(|url| !seen.contains(*url) && !pending.contains(*url))
                .cloned()
                .collect()
        };

        // Rate-limit gate: drop excess submissions on the floor with a WARN.
        // The mTLS verifier must remain non-blocking, so we use the
        // synchronous `check()` API and never await here.
        for url in candidates {
            if self.discovery_limiter.check().is_err() {
                tracing::warn!(
                    url = %url,
                    "discovery_rate_limited: dropped CDP URL beyond per-minute cap (will be retried on next handshake observing this URL)"
                );
                continue;
            }
            if self.discover_tx.send(url.clone()).is_err() {
                // Receiver gone (shutdown). Do NOT mark pending so the
                // URL can be retried after a reload / restart.
                tracing::debug!(
                    url = %url,
                    "discover channel closed; dropping CDP URL without marking pending"
                );
                continue;
            }
            // Queued for fetch. Mark pending (not seen) so concurrent
            // handshakes do not re-enqueue it while the fetch is in flight.
            let mut guard = self
                .pending_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            if guard.len() >= self.config.crl_max_seen_urls {
                self.warn_cap_exceeded_throttled("pending_urls");
                break;
            }
            guard.insert(url);
        }

        if self.config.crl_deny_on_unavailable {
            let cached = self
                .cached_urls
                .lock()
                .ok()
                .map(|guard| guard.clone())
                .unwrap_or_default();
            let relevant_urls = if self.config.crl_end_entity_only {
                end_entity_urls
            } else {
                all_urls.as_slice()
            };
            return !relevant_urls.is_empty()
                && relevant_urls.iter().all(|url| !cached.contains(url));
        }

        false
    }

    /// Test helper for constructing a CRL set from in-memory CRLs.
    ///
    /// # Errors
    ///
    /// Returns an error if the verifier cannot be built from the provided CRLs.
    #[doc(hidden)]
    pub fn __test_with_prepopulated_crls(
        roots: Arc<RootCertStore>,
        config: MtlsConfig,
        prefilled_crls: Vec<CertificateRevocationListDer<'static>>,
    ) -> Result<Arc<Self>, RmcpServerKitError> {
        let (discover_tx, discover_rx) = mpsc::unbounded_channel();
        drop(discover_rx);

        let mut initial_cache = HashMap::new();
        for (index, der) in prefilled_crls.into_iter().enumerate() {
            let source_url = format!("memory://crl/{index}");
            let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;
            initial_cache.insert(
                source_url.clone(),
                CachedCrl {
                    der,
                    this_update,
                    next_update,
                    fetched_at: SystemTime::now(),
                    source_url,
                },
            );
        }

        Self::new(roots, config, discover_tx, initial_cache)
    }

    /// Test-only: same as [`Self::__test_with_prepopulated_crls`] but
    /// returns the discover-channel receiver to the caller so the
    /// background channel `send`s succeed (the receiver stays alive
    /// for the duration of the test). Required by the B2 dedup
    /// regression test, which must observe URLs being committed to
    /// `seen_urls` after a successful limiter+send sequence. Not part
    /// of the public API.
    ///
    /// # Errors
    ///
    /// Returns an error if the verifier cannot be built from the provided CRLs.
    #[doc(hidden)]
    pub fn __test_with_kept_receiver(
        roots: Arc<RootCertStore>,
        config: MtlsConfig,
        prefilled_crls: Vec<CertificateRevocationListDer<'static>>,
    ) -> Result<(Arc<Self>, mpsc::UnboundedReceiver<String>), RmcpServerKitError> {
        let (discover_tx, discover_rx) = mpsc::unbounded_channel();

        let mut initial_cache = HashMap::new();
        for (index, der) in prefilled_crls.into_iter().enumerate() {
            let source_url = format!("memory://crl/{index}");
            let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;
            initial_cache.insert(
                source_url.clone(),
                CachedCrl {
                    der,
                    this_update,
                    next_update,
                    fetched_at: SystemTime::now(),
                    source_url,
                },
            );
        }

        let crl_set = Self::new(roots, config, discover_tx, initial_cache)?;
        Ok((crl_set, discover_rx))
    }

    /// Test-only: directly invoke the discovery rate-limiter on a batch of URLs
    /// and return `(accepted, dropped)`. Bypasses the dedup `seen_urls` set so
    /// callers can deterministically saturate the limiter; mutates the limiter
    /// state in place. Not part of the public API.
    #[doc(hidden)]
    pub fn __test_check_discovery_rate(&self, urls: &[String]) -> (usize, usize) {
        let mut accepted = 0usize;
        let mut dropped = 0usize;
        for url in urls {
            if self.discovery_limiter.check().is_ok() {
                let _ = self.discover_tx.send(url.clone());
                accepted += 1;
            } else {
                dropped += 1;
            }
        }
        (accepted, dropped)
    }

    /// Test-only: invoke the real `note_discovered_urls` so dedup + rate-limit
    /// + cached-fallback paths are all exercised. Returns the `missing_cached`
    /// flag the production verifier uses to decide whether to fail the handshake.
    ///
    /// When no receiver is attached (the usual unit-test setup), the send fails
    /// and production correctly records nothing, so this mirrors the admission
    /// bookkeeping by marking the URL **pending** — matching what a live
    /// refresher would observe between enqueue and fetch.
    #[doc(hidden)]
    pub fn __test_note_discovered_urls(&self, urls: &[String]) -> bool {
        let missing_cached = self.note_discovered_urls(urls, &[]);
        if self.discover_tx.is_closed() {
            let already_seen: HashSet<String> = {
                let seen = self
                    .seen_urls
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                urls.iter()
                    .filter(|url| seen.contains(*url))
                    .cloned()
                    .collect()
            };
            let mut pending = self
                .pending_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            for url in urls {
                if already_seen.contains(url) || pending.contains(url) {
                    continue;
                }
                if pending.len() >= self.config.crl_max_seen_urls {
                    self.warn_cap_exceeded_throttled("pending_urls");
                    break;
                }
                pending.insert(url.clone());
            }
        }
        missing_cached
    }

    /// Test-only: invoke the real precheck with separate end-entity and
    /// intermediate CDP sets.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_note_discovered_urls_by_cert(
        &self,
        end_entity_urls: &[String],
        intermediate_urls: &[String],
    ) -> bool {
        self.note_discovered_urls(end_entity_urls, intermediate_urls)
    }

    /// Test-only: report whether a URL is currently suppressed from
    /// re-discovery — i.e. present in EITHER dedup state.
    ///
    /// This is the property callers actually care about: "will a future
    /// handshake re-enqueue this URL?". Use
    /// [`Self::__test_is_permanently_seen`] when the distinction between
    /// in-flight and confirmed-cached matters.
    #[doc(hidden)]
    pub fn __test_is_seen(&self, url: &str) -> bool {
        let in_seen = {
            let seen = self
                .seen_urls
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            seen.contains(url)
        };
        if in_seen {
            return true;
        }
        let pending = self
            .pending_urls
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        pending.contains(url)
    }

    /// Test-only: report whether a URL reached the PERMANENT dedup set,
    /// which happens only after its CRL is confirmed present in the cache.
    /// A URL that was merely queued, or whose fetch failed, is not counted.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_is_permanently_seen(&self, url: &str) -> bool {
        let seen = self
            .seen_urls
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        seen.contains(url)
    }

    /// Test-only: drive the post-fetch bookkeeping without performing HTTP.
    /// `admitted` mirrors [`Self::fetch_and_store_url`]'s return value:
    /// `true` when the CRL landed in the cache, `false` when the fetch
    /// failed or the cache cap refused it.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_settle_pending(&self, url: &str, admitted: bool) {
        if admitted {
            self.promote_pending_to_seen(url);
        } else {
            self.clear_pending(url);
        }
    }

    /// Test-only: current count of host semaphores. Used by
    /// `tests/crl_map_bounds.rs` to assert the cap is enforced.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_host_semaphore_count(&self) -> usize {
        self.host_semaphores
            .try_lock()
            .map_or(0, |guard| guard.len())
    }

    /// Test-only: current number of entries in the CRL cache.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_cache_len(&self) -> usize {
        self.cache.try_read().map_or(0, |guard| guard.len())
    }

    /// Test-only: whether a specific URL is currently cached.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_cache_contains(&self, url: &str) -> bool {
        self.cache
            .try_read()
            .is_ok_and(|guard| guard.contains_key(url))
    }

    /// Test-only: whether a URL is advertised to the fail-closed precheck as
    /// present in the live verifier.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub fn __test_cached_url_contains(&self, url: &str) -> bool {
        self.cached_urls
            .lock()
            .is_ok_and(|guard| guard.contains(url))
    }

    /// Test-only: triggers the request-hot-path fetch path for `url`
    /// WITHOUT going through the TLS handshake. Returns any error the
    /// host-semaphore cap check produces. A network-unreachable
    /// failure for the fetch itself is treated as `Ok(())` (test only
    /// cares about the cap; real tests use mock hosts that won't
    /// resolve — the cap must fire BEFORE network I/O).
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub async fn __test_trigger_fetch(&self, url: &str) -> Result<(), RmcpServerKitError> {
        if let Err(error) = gated_fetch(
            &self.client,
            &self.global_fetch_sem,
            &self.host_semaphores,
            url,
            self.config.crl_allow_http,
            self.max_response_bytes,
            self.config.crl_max_host_semaphores,
        )
        .await
        {
            if error
                .to_string()
                .contains("crl_host_semaphore_cap_exceeded")
            {
                Err(error)
            } else {
                Ok(())
            }
        } else {
            Ok(())
        }
    }

    /// Test-only: directly insert `cached` under `url` into both
    /// `cache` and `cached_urls`, bypassing HTTP. Does NOT enforce
    /// `crl_max_cache_entries` when called pre-cap — the test uses it
    /// to stage preconditions. For cap-breach coverage, tests invoke
    /// the real production insertion path.
    ///
    /// Wait — the `cache_hard_cap_drops_newest` test DOES use this
    /// helper to assert the cap fires. Therefore this helper MUST
    /// enforce the hard cap (silent drop with warn!) the same way the
    /// production code does. The helper is a thin wrapper around the
    /// same internal insertion fn the production path uses.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub async fn __test_insert_cache(&self, url: &str, cached: CachedCrl) {
        let _ = self
            .commit_cache_update_atomically(vec![(url.to_owned(), cached)], &[])
            .await;
    }

    /// Test-only: direct cache insertion that returns verifier rebuild errors.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub async fn __test_try_insert_cache(
        &self,
        url: &str,
        cached: CachedCrl,
    ) -> Result<bool, RmcpServerKitError> {
        self.commit_cache_update_atomically(vec![(url.to_owned(), cached)], &[])
            .await
    }

    /// Test-only: replace a cache entry without rebuilding the verifier.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub async fn __test_replace_cache_entry_unverified(&self, url: &str, cached: CachedCrl) {
        let mut cache = self.cache.write().await;
        cache.insert(url.to_owned(), cached);
    }

    /// Test-only: trigger a refresh cycle for a single URL. Exercises
    /// the same stale-grace / fetch-failure path as `refresh_urls()`.
    /// Returns the refresh error (if any) — most tests ignore it
    /// because they assert post-state, not the transient error.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    pub async fn __test_trigger_refresh_url(&self, url: &str) -> Result<(), RmcpServerKitError> {
        self.refresh_urls(vec![url.to_owned()]).await
    }

    async fn fetch_url_results(
        &self,
        urls: Vec<String>,
    ) -> Vec<(String, Result<CachedCrl, RmcpServerKitError>)> {
        let mut tasks = JoinSet::new();
        for url in urls {
            let client = self.client.clone();
            let global_sem = Arc::clone(&self.global_fetch_sem);
            let host_map = Arc::clone(&self.host_semaphores);
            let allow_http = self.config.crl_allow_http;
            let max_bytes = self.max_response_bytes;
            let max_host_semaphores = self.config.crl_max_host_semaphores;
            tasks.spawn(async move {
                let result = gated_fetch(
                    &client,
                    &global_sem,
                    &host_map,
                    &url,
                    allow_http,
                    max_bytes,
                    max_host_semaphores,
                )
                .await;
                (url, result)
            });
        }

        let mut results = Vec::new();
        while let Some(joined) = tasks.join_next().await {
            match joined {
                Ok(result) => results.push(result),
                Err(error) => {
                    tracing::warn!(error = %error, "CRL refresh task join failed");
                }
            }
        }

        results
    }
}

#[cfg(any(test, feature = "test-helpers"))]
const SYNTHETIC_TEST_CRL_DER: &[u8] = &[
    48, 129, 199, 48, 110, 2, 1, 1, 48, 10, 6, 8, 42, 134, 72, 206, 61, 4, 3, 2, 48, 14, 49, 12,
    48, 10, 6, 3, 85, 4, 3, 12, 3, 99, 114, 108, 23, 13, 50, 54, 48, 49, 48, 49, 48, 48, 48, 48,
    48, 48, 90, 23, 13, 50, 55, 48, 49, 48, 49, 48, 48, 48, 48, 48, 48, 90, 160, 47, 48, 45, 48,
    31, 6, 3, 85, 29, 35, 4, 24, 48, 22, 128, 20, 14, 62, 48, 146, 7, 182, 179, 215, 90, 226, 214,
    90, 201, 83, 149, 116, 34, 31, 26, 255, 48, 10, 6, 3, 85, 29, 20, 4, 3, 2, 1, 1, 48, 10, 6, 8,
    42, 134, 72, 206, 61, 4, 3, 2, 3, 73, 0, 48, 70, 2, 33, 0, 250, 240, 103, 87, 60, 78, 208, 171,
    184, 206, 117, 134, 236, 234, 53, 115, 122, 90, 64, 217, 146, 27, 32, 103, 170, 222, 240, 159,
    137, 187, 116, 6, 2, 33, 0, 188, 23, 204, 232, 130, 84, 135, 249, 43, 208, 224, 220, 202, 57,
    98, 140, 4, 251, 148, 189, 105, 68, 105, 40, 53, 180, 208, 38, 193, 120, 118, 100,
];

impl CachedCrl {
    /// Test-only: synthesize a cache entry that looks valid, `next_update`
    /// = now + 24h. Fields used only to populate the HashMap — the bytes
    /// are a minimal CRL-shape that won't be parsed by tests.
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    #[must_use]
    pub fn __test_synthetic(now: SystemTime) -> Self {
        Self {
            der: CertificateRevocationListDer::from(SYNTHETIC_TEST_CRL_DER.to_vec()),
            this_update: now,
            next_update: now.checked_add(Duration::from_hours(24)),
            fetched_at: now,
            source_url: "test://synthetic".to_owned(),
        }
    }

    /// Test-only: synthesize a STALE cache entry (`next_update` in the
    /// deep past so `is_stale_beyond_grace` fires with any sensible
    /// `crl_stale_grace`).
    #[cfg(any(test, feature = "test-helpers"))]
    #[doc(hidden)]
    #[must_use]
    pub fn __test_stale(reference_past: SystemTime) -> Self {
        Self {
            der: CertificateRevocationListDer::from(vec![0x30, 0x00]),
            this_update: reference_past,
            next_update: Some(reference_past),
            fetched_at: reference_past,
            source_url: "test://stale".to_owned(),
        }
    }
}

/// Stable outer verifier that delegates all TLS verification behavior to the
/// atomically swappable inner verifier.
pub struct DynamicClientCertVerifier {
    inner: Arc<CrlSet>,
    dn_subjects: Vec<DistinguishedName>,
}

impl DynamicClientCertVerifier {
    /// Construct a new dynamic verifier from a shared [`CrlSet`].
    #[must_use]
    pub fn new(inner: Arc<CrlSet>) -> Self {
        Self {
            dn_subjects: inner.roots.subjects(),
            inner,
        }
    }
}

impl std::fmt::Debug for DynamicClientCertVerifier {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DynamicClientCertVerifier")
            .field("dn_subjects_len", &self.dn_subjects.len())
            .finish_non_exhaustive()
    }
}

impl ClientCertVerifier for DynamicClientCertVerifier {
    fn offer_client_auth(&self) -> bool {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.offer_client_auth()
    }

    fn client_auth_mandatory(&self) -> bool {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.client_auth_mandatory()
    }

    fn root_hint_subjects(&self) -> &[DistinguishedName] {
        &self.dn_subjects
    }

    fn verify_client_cert(
        &self,
        end_entity: &CertificateDer<'_>,
        intermediates: &[CertificateDer<'_>],
        now: UnixTime,
    ) -> Result<ClientCertVerified, TlsError> {
        // SECURITY: extracting CDP URLs from an unverified client cert
        // here is intentional. No HTTP happens on this path -- the call
        // to `note_discovered_urls` only enqueues onto a bounded,
        // rate-limited channel. The actual fetch runs off-path in
        // `run_crl_refresher` and is gated by SSRF screening
        // (`src/ssrf.rs`), body-size cap, deadline, and the
        // `crl_allow_http` policy. CRLs are CA-signed (RFC 5280 §5), so
        // http(s) CDP URLs are protocol design, not an SSRF sink. The
        // discovery must happen BEFORE delegating to the inner verifier
        // so `crl_deny_on_unavailable = true` can fail-closed on a
        // never-fetched CDP. Do NOT reorder.
        let mut end_entity_urls =
            extract_cdp_urls(end_entity.as_ref(), self.inner.config.crl_allow_http);
        end_entity_urls.sort();
        end_entity_urls.dedup();

        let mut intermediate_urls = Vec::new();
        for intermediate in intermediates {
            intermediate_urls.extend(extract_cdp_urls(
                intermediate.as_ref(),
                self.inner.config.crl_allow_http,
            ));
        }
        intermediate_urls.sort();
        intermediate_urls.dedup();

        if self
            .inner
            .note_discovered_urls(&end_entity_urls, &intermediate_urls)
        {
            return Err(TlsError::General(
                "client certificate revocation status unavailable".to_owned(),
            ));
        }

        let verifier = self.inner.inner_verifier.load();
        verifier
            .0
            .verify_client_cert(end_entity, intermediates, now)
    }

    fn verify_tls12_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, TlsError> {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.verify_tls12_signature(message, cert, dss)
    }

    fn verify_tls13_signature(
        &self,
        message: &[u8],
        cert: &CertificateDer<'_>,
        dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, TlsError> {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.verify_tls13_signature(message, cert, dss)
    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.supported_verify_schemes()
    }

    fn requires_raw_public_keys(&self) -> bool {
        let verifier = self.inner.inner_verifier.load();
        verifier.0.requires_raw_public_keys()
    }
}

/// Extract CRL Distribution Point URLs from a DER-encoded certificate.
///
/// URLs are validated with `url::Url::parse` (case-insensitive scheme handling)
/// and filtered through an internal scheme guard. Malformed URLs, URLs
/// using disallowed schemes, and URLs carrying embedded credentials
/// (userinfo) are silently dropped. SSRF defenses against private
/// IP literals and metadata endpoints are applied later, at fetch time, after
/// DNS resolution.
#[must_use]
pub fn extract_cdp_urls(cert_der: &[u8], allow_http: bool) -> Vec<String> {
    let Ok((_, cert)) = X509Certificate::from_der(cert_der) else {
        return Vec::new();
    };

    let mut urls = Vec::new();
    for ext in cert.extensions() {
        if let ParsedExtension::CRLDistributionPoints(cdps) = ext.parsed_extension() {
            for point in cdps.iter() {
                if let Some(DistributionPointName::FullName(names)) = &point.distribution_point {
                    for name in names {
                        if let GeneralName::URI(uri) = name {
                            let raw = *uri;
                            let Ok(parsed) = Url::parse(raw) else {
                                // `?raw` (Debug) escapes control characters the
                                // failed parse may have left in this
                                // attacker-supplied string.
                                tracing::debug!(url = ?raw, "CDP URL parse failed; dropped");
                                continue;
                            };
                            if let Err(reason) = check_scheme(&parsed, allow_http) {
                                tracing::debug!(
                                    url = %sanitized_url_for_log(&parsed),
                                    reason,
                                    "CDP URL rejected by scheme guard; dropped"
                                );
                                continue;
                            }
                            urls.push(parsed.into());
                        }
                    }
                }
            }
        }
    }

    urls
}

/// Bootstrap the CRL cache by extracting CDP URLs from the CA chain and
/// fetching any reachable CRLs with a 10-second total deadline.
///
/// # Errors
///
/// Returns an error if the initial verifier cannot be built.
#[allow(
    clippy::cognitive_complexity,
    reason = "bootstrap coordinates timeout, parallel fetches, and partial-cache recovery"
)]
pub async fn bootstrap_fetch(
    roots: Arc<RootCertStore>,
    ca_certs: &[CertificateDer<'static>],
    config: MtlsConfig,
) -> Result<(Arc<CrlSet>, mpsc::UnboundedReceiver<String>), RmcpServerKitError> {
    let (discover_tx, discover_rx) = mpsc::unbounded_channel();

    let mut urls = ca_certs
        .iter()
        .flat_map(|cert| extract_cdp_urls(cert.as_ref(), config.crl_allow_http))
        .collect::<Vec<_>>();
    urls.sort();
    urls.dedup();

    // M-H2: same SSRF resolver hardening as CrlSet::new -- bootstrap
    // fetches the same attacker-controlled CDP URLs, just earlier in
    // the lifecycle.
    let bootstrap_allowlist = Arc::new(crate::ssrf::CompiledSsrfAllowlist::default());
    let bootstrap_resolver: Arc<dyn reqwest::dns::Resolve> =
        Arc::new(crate::ssrf_resolver::SsrfScreeningResolver::new(
            Arc::clone(&bootstrap_allowlist),
            #[cfg(any(test, feature = "test-helpers"))]
            Arc::new(std::sync::atomic::AtomicBool::new(false)),
            #[cfg(not(any(test, feature = "test-helpers")))]
            (),
        ));

    let client = reqwest::Client::builder()
        // M-H2/N1: see oauth.rs::OauthHttpClient::build for rationale.
        .no_proxy()
        .dns_resolver(Arc::clone(&bootstrap_resolver))
        .timeout(config.crl_fetch_timeout)
        .connect_timeout(CRL_CONNECT_TIMEOUT)
        .tcp_keepalive(None)
        .redirect(reqwest::redirect::Policy::none())
        .user_agent(format!("rmcp-server-kit/{}", env!("CARGO_PKG_VERSION")))
        .build()
        .map_err(|error| RmcpServerKitError::Startup(format!("CRL HTTP client init: {error}")))?;

    // Bootstrap shares the same global concurrency + per-host cap as the
    // hot-path verifier so a maliciously broad CA chain cannot overwhelm
    // the network at startup.
    let bootstrap_concurrency = config.crl_max_concurrent_fetches.max(1);
    let global_sem = Arc::new(Semaphore::new(bootstrap_concurrency));
    let host_semaphores = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
    let allow_http = config.crl_allow_http;
    let max_bytes = config.crl_max_response_bytes;
    let max_host_semaphores = config.crl_max_host_semaphores;

    let mut initial_cache = HashMap::new();
    let mut tasks = JoinSet::new();
    for url in &urls {
        let client = client.clone();
        let url = url.clone();
        let global_sem = Arc::clone(&global_sem);
        let host_semaphores = Arc::clone(&host_semaphores);
        tasks.spawn(async move {
            let result = gated_fetch(
                &client,
                &global_sem,
                &host_semaphores,
                &url,
                allow_http,
                max_bytes,
                max_host_semaphores,
            )
            .await;
            (url, result)
        });
    }

    let timeout: Sleep = tokio::time::sleep(BOOTSTRAP_TIMEOUT);
    tokio::pin!(timeout);

    while !tasks.is_empty() {
        // cancel-safe: pinned Sleep and JoinSet::join_next are cancel-safe
        // (tokio docs); on timeout the loop breaks and dropping the JoinSet
        // aborts remaining fetches — the intended deadline behavior.
        tokio::select! {
            () = &mut timeout => {
                tracing::warn!("CRL bootstrap timed out after {:?}", BOOTSTRAP_TIMEOUT);
                break;
            }
            maybe_joined = tasks.join_next() => {
                let Some(joined) = maybe_joined else {
                    break;
                };
                match joined {
                    Ok((url, Ok(cached))) => {
                        initial_cache.insert(url, cached);
                    }
                    Ok((url, Err(error))) => {
                        tracing::warn!(url = %url, error = %error, "CRL bootstrap fetch failed");
                    }
                    Err(error) => {
                        tracing::warn!(error = %error, "CRL bootstrap task join failed");
                    }
                }
            }
        }
    }

    let set = CrlSet::new(roots, config, discover_tx, initial_cache)?;
    Ok((set, discover_rx))
}

/// Run the CRL refresher loop until shutdown.
#[allow(
    clippy::cognitive_complexity,
    reason = "refresher loop intentionally handles shutdown, timer, and discovery in one select"
)]
pub async fn run_crl_refresher(
    set: Arc<CrlSet>,
    mut discover_rx: mpsc::UnboundedReceiver<String>,
    shutdown: CancellationToken,
) {
    let mut refresh_sleep = schedule_next_refresh(&set).await;

    loop {
        // cancel-safe: CancellationToken::cancelled, pinned &mut Sleep, and
        // mpsc::UnboundedReceiver::recv are all cancel-safe (tokio docs);
        // refresh work happens inside arm bodies, never in the raced futures.
        tokio::select! {
            () = shutdown.cancelled() => {
                break;
            }
            () = &mut refresh_sleep => {
                if let Err(error) = set.refresh_due_urls().await {
                    tracing::warn!(error = %error, "CRL periodic refresh failed");
                }
                refresh_sleep = schedule_next_refresh(&set).await;
            }
            maybe_url = discover_rx.recv() => {
                let Some(url) = maybe_url else {
                    break;
                };
                match set.fetch_and_store_url(url.clone()).await {
                    // Cached: safe to suppress this URL permanently.
                    Ok(true) => set.promote_pending_to_seen(&url),
                    // Fetched but refused by the cache cap. Clear the
                    // in-flight marker so a later handshake can retry;
                    // suppressing it here would disable revocation for
                    // this CDP even though no CRL was ever cached.
                    Ok(false) => {
                        set.clear_pending(&url);
                        tracing::warn!(
                            url = %url,
                            "CRL fetched but not admitted to cache (cap reached); will retry on a later handshake"
                        );
                    }
                    Err(error) => {
                        set.clear_pending(&url);
                        tracing::warn!(
                            url = %url,
                            error = %error,
                            "CRL discovery fetch failed; will retry on a later handshake"
                        );
                    }
                }
                refresh_sleep = schedule_next_refresh(&set).await;
            }
        }
    }
}

/// Rebuild the inner rustls verifier from the current CRL cache.
///
/// # Errors
///
/// Returns an error if rustls rejects the verifier configuration.
pub fn rebuild_verifier<S: std::hash::BuildHasher>(
    roots: &Arc<RootCertStore>,
    config: &MtlsConfig,
    cache: &HashMap<String, CachedCrl, S>,
) -> Result<Arc<dyn ClientCertVerifier>, RmcpServerKitError> {
    let mut builder = WebPkiClientVerifier::builder(Arc::clone(roots));

    if !cache.is_empty() {
        let crls = cache
            .values()
            .map(|cached| cached.der.clone())
            .collect::<Vec<_>>();
        builder = builder.with_crls(crls);
    }
    if config.crl_end_entity_only {
        builder = builder.only_check_end_entity_revocation();
    }
    if !config.crl_deny_on_unavailable {
        builder = builder.allow_unknown_revocation_status();
    }
    if config.crl_enforce_expiration {
        builder = builder.enforce_revocation_expiration();
    }
    if !config.required {
        builder = builder.allow_unauthenticated();
    }

    builder
        .build()
        .map_err(|error| RmcpServerKitError::Tls(format!("mTLS verifier error: {error}")))
}

/// Parse `thisUpdate` and `nextUpdate` metadata from a DER-encoded CRL.
///
/// # Errors
///
/// Returns an error if the CRL cannot be parsed.
pub fn parse_crl_metadata(
    der: &[u8],
) -> Result<(SystemTime, Option<SystemTime>), RmcpServerKitError> {
    let (_, crl) = CertificateRevocationList::from_der(der)
        .map_err(|error| RmcpServerKitError::Tls(format!("invalid CRL DER: {error:?}")))?;

    Ok((
        asn1_time_to_system_time(crl.last_update()),
        crl.next_update().map(asn1_time_to_system_time),
    ))
}

async fn schedule_next_refresh(set: &CrlSet) -> Pin<Box<Sleep>> {
    let duration = next_refresh_delay(set).await;
    boxed_sleep(duration)
}

fn boxed_sleep(duration: Duration) -> Pin<Box<Sleep>> {
    Box::pin(tokio::time::sleep_until(Instant::now() + duration))
}

async fn next_refresh_delay(set: &CrlSet) -> Duration {
    if let Some(interval) = set.config.crl_refresh_interval {
        return clamp_refresh(interval);
    }

    let now = SystemTime::now();
    let cache = set.cache.read().await;
    let mut next = MAX_AUTO_REFRESH;

    for cached in cache.values() {
        if let Some(next_update) = cached.next_update {
            let duration = next_update.duration_since(now).unwrap_or(Duration::ZERO);
            next = next.min(clamp_refresh(duration));
        }
    }
    drop(cache);

    next
}

/// Get-or-insert the per-host fetch semaphore for `host_key`.
///
/// When the map is at `max_host_semaphores`, idle entries (no in-flight
/// fetch) are evicted before rejecting, so the cap only fails when `max`
/// distinct hosts are *concurrently* fetching — it is never a permanent
/// lockout. Every clone of a host semaphore is created while holding the
/// map lock, and a clone outlives the critical section only while a fetch
/// is in flight, so an entry with `Arc::strong_count == 1` is provably
/// idle and safe to drop.
fn acquire_host_semaphore(
    map: &mut HashMap<String, Arc<Semaphore>>,
    host_key: &str,
    max_host_semaphores: usize,
) -> Result<Arc<Semaphore>, RmcpServerKitError> {
    if !map.contains_key(host_key) {
        if map.len() >= max_host_semaphores {
            // Self-heal: drop semaphores with no in-flight fetch.
            map.retain(|_, semaphore| Arc::strong_count(semaphore) > 1);
        }
        if map.len() >= max_host_semaphores {
            return Err(RmcpServerKitError::Config(
                "crl_host_semaphore_cap_exceeded: too many distinct CRL hosts in flight".to_owned(),
            ));
        }
        map.insert(host_key.to_owned(), Arc::new(Semaphore::new(1)));
    }
    match map.get(host_key) {
        Some(semaphore) => Ok(Arc::clone(semaphore)),
        None => Err(RmcpServerKitError::Tls(
            "CRL host semaphore missing after insertion".to_owned(),
        )),
    }
}

/// Fetch a single CRL URL through the global + per-host concurrency caps.
///
/// `global_sem` caps total simultaneous CRL fetches process-wide.
/// `host_semaphores` ensures at most one in-flight fetch per origin host
/// (an SSRF amplification defense); at the host cap, idle entries are
/// evicted on demand. Both permits are dropped when the returned future
/// completes (whether `Ok` or `Err`).
async fn gated_fetch(
    client: &reqwest::Client,
    global_sem: &Arc<Semaphore>,
    host_semaphores: &Arc<tokio::sync::Mutex<HashMap<String, Arc<Semaphore>>>>,
    url: &str,
    allow_http: bool,
    max_bytes: u64,
    max_host_semaphores: usize,
) -> Result<CachedCrl, RmcpServerKitError> {
    let host_key = Url::parse(url)
        .ok()
        .and_then(|u| u.host_str().map(str::to_owned))
        .unwrap_or_else(|| url.to_owned());

    let host_sem = {
        let mut map = host_semaphores.lock().await;
        acquire_host_semaphore(&mut map, &host_key, max_host_semaphores)?
    };

    let _global_permit = Arc::clone(global_sem)
        .acquire_owned()
        .await
        .map_err(|error| {
            RmcpServerKitError::Tls(format!("CRL global semaphore closed: {error}"))
        })?;
    let _host_permit = host_sem
        .acquire_owned()
        .await
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL host semaphore closed: {error}")))?;

    fetch_crl(client, url, allow_http, max_bytes).await
}

async fn fetch_crl(
    client: &reqwest::Client,
    url: &str,
    allow_http: bool,
    max_bytes: u64,
) -> Result<CachedCrl, RmcpServerKitError> {
    let parsed = Url::parse(url)
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL URL parse {url}: {error}")))?;

    if let Err(reason) = check_scheme(&parsed, allow_http) {
        // Sanitized: the gate must not echo what it rejects (the URL may
        // carry userinfo credentials — the very thing being refused).
        let sanitized = sanitized_url_for_log(&parsed);
        tracing::warn!(url = %sanitized, reason, "CRL fetch denied: scheme");
        return Err(RmcpServerKitError::Tls(format!(
            "CRL scheme rejected ({reason}): {sanitized}"
        )));
    }

    let host = parsed
        .host_str()
        .ok_or_else(|| RmcpServerKitError::Tls(format!("CRL URL has no host: {url}")))?;
    let port = parsed
        .port_or_known_default()
        .ok_or_else(|| RmcpServerKitError::Tls(format!("CRL URL has no known port: {url}")))?;

    let addrs = lookup_host((host, port))
        .await
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL DNS resolution {url}: {error}")))?;

    let mut any_addr = false;
    for addr in addrs {
        any_addr = true;
        if let Some(reason) = ip_block_reason(addr.ip()) {
            tracing::warn!(
                url = %url,
                resolved_ip = %addr.ip(),
                reason,
                "CRL fetch denied: blocked IP"
            );
            return Err(RmcpServerKitError::Tls(format!(
                "CRL host resolved to blocked IP ({reason}): {url}"
            )));
        }
    }
    if !any_addr {
        return Err(RmcpServerKitError::Tls(format!(
            "CRL DNS resolution returned no addresses: {url}"
        )));
    }

    let mut response = client
        .get(url)
        .send()
        .await
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL fetch {url}: {error}")))?
        .error_for_status()
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL fetch {url}: {error}")))?;

    // Enforce body cap by streaming chunk-by-chunk; a malicious or
    // misconfigured server cannot allocate more than `max_bytes` of memory.
    let initial_capacity = usize::try_from(max_bytes.min(64 * 1024)).unwrap_or(64 * 1024);
    let mut body: Vec<u8> = Vec::with_capacity(initial_capacity);
    while let Some(chunk) = response
        .chunk()
        .await
        .map_err(|error| RmcpServerKitError::Tls(format!("CRL read {url}: {error}")))?
    {
        let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
        let body_len = u64::try_from(body.len()).unwrap_or(u64::MAX);
        if body_len.saturating_add(chunk_len) > max_bytes {
            return Err(RmcpServerKitError::Tls(format!(
                "CRL body exceeded cap of {max_bytes} bytes: {url}"
            )));
        }
        body.extend_from_slice(&chunk);
    }

    let der = CertificateRevocationListDer::from(body);
    let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;

    Ok(CachedCrl {
        der,
        this_update,
        next_update,
        fetched_at: SystemTime::now(),
        source_url: url.to_owned(),
    })
}

fn should_refresh_cached(
    cached: &CachedCrl,
    now: SystemTime,
    fixed_interval: Option<Duration>,
) -> bool {
    if let Some(interval) = fixed_interval {
        return cached
            .fetched_at
            .checked_add(clamp_refresh(interval))
            .is_none_or(|deadline| now >= deadline);
    }

    cached
        .next_update
        .is_none_or(|next_update| now >= next_update)
}

fn clamp_refresh(duration: Duration) -> Duration {
    duration.clamp(MIN_AUTO_REFRESH, MAX_AUTO_REFRESH)
}

/// 9999-12-31T23:59:59Z — the maximum instant expressible as an ASN.1
/// GeneralizedTime (four-digit year). Used to clamp absurd positive
/// timestamps before converting to [`SystemTime`].
const MAX_ASN1_TIMESTAMP_SECS: u64 = 253_402_300_799;

/// Convert an ASN.1 time to [`SystemTime`] without ever panicking.
///
/// CRL metadata is parsed from raw fetched bytes *before* signature
/// validation, so timestamps are attacker-controlled. Platform
/// `SystemTime` ranges differ (Windows cannot represent pre-1601);
/// unrepresentable values are clamped toward [`UNIX_EPOCH`], which is the
/// safe direction: it can only make a CRL look *older* (forcing an
/// eager refresh), never fresher.
fn asn1_time_to_system_time(time: x509_parser::time::ASN1Time) -> SystemTime {
    let timestamp = time.timestamp();
    if timestamp >= 0 {
        let seconds = u64::try_from(timestamp)
            .unwrap_or(0)
            .min(MAX_ASN1_TIMESTAMP_SECS);
        UNIX_EPOCH
            .checked_add(Duration::from_secs(seconds))
            .unwrap_or(UNIX_EPOCH)
    } else {
        UNIX_EPOCH
            .checked_sub(Duration::from_secs(timestamp.unsigned_abs()))
            .unwrap_or(UNIX_EPOCH)
    }
}

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

    fn asn1(timestamp: i64) -> x509_parser::time::ASN1Time {
        x509_parser::time::ASN1Time::from_timestamp(timestamp).expect("valid ASN.1 timestamp")
    }

    /// The userinfo gate fires before DNS resolution (no network needed)
    /// and the surfaced error must not echo the rejected credentials.
    #[tokio::test]
    async fn fetch_crl_rejects_userinfo_without_echoing_credentials() {
        // reqwest with `rustls-no-provider` requires a process-wide crypto
        // provider before any Client is built (same pattern as the
        // transport/oauth test suites).
        let _ = rustls::crypto::ring::default_provider().install_default();
        let client = reqwest::Client::new();
        let err = fetch_crl(&client, "https://u:p@crl.example/ca.crl", false, 1024)
            .await
            .expect_err("userinfo-bearing CRL URL must be rejected");
        let rendered = err.to_string();
        assert!(
            rendered.contains("userinfo_forbidden"),
            "error must carry the rejection reason: {rendered}"
        );
        assert!(
            !rendered.contains("u:p"),
            "error must not echo the rejected credentials: {rendered}"
        );
    }

    /// `extract_cdp_urls`'s scheme/userinfo guard reuses the same gate;
    /// the sanitizer keeps credentials out of its debug logging too.
    #[test]
    fn sanitizer_used_by_rejection_sites_strips_credentials() {
        let parsed = Url::parse("https://u:p@crl.example/ca.crl").expect("parse");
        let sanitized = sanitized_url_for_log(&parsed);
        assert_eq!(sanitized, "https://crl.example");
        assert!(!sanitized.contains("u:p"));
    }

    #[test]
    fn asn1_time_clamps_unrepresentable_timestamps() {
        // Year 1500 — pre-1601, NOT representable by Windows `SystemTime`.
        // Pre-fix this panicked on Windows; now it must return a value no
        // later than the epoch on every platform (clamped to UNIX_EPOCH on
        // Windows, the real instant on platforms that can represent it).
        let year_1500 = asn1_time_to_system_time(asn1(-14_831_769_600));
        assert!(year_1500 <= UNIX_EPOCH);
        #[cfg(windows)]
        assert_eq!(year_1500, UNIX_EPOCH);

        // 1601-01-01T00:00:00Z — the exact Windows epoch boundary, which IS
        // representable everywhere. No clamp, no panic.
        let year_1601 = asn1_time_to_system_time(asn1(-11_644_473_600));
        assert!(year_1601 <= UNIX_EPOCH);

        // Mildly negative (pre-1970) stays at-or-before the epoch.
        assert!(asn1_time_to_system_time(asn1(-2)) <= UNIX_EPOCH);

        // Normal positive timestamps round-trip exactly.
        assert_eq!(
            asn1_time_to_system_time(asn1(1_700_000_000)),
            UNIX_EPOCH + Duration::from_secs(1_700_000_000)
        );

        // The ASN.1 maximum (9999-12-31) is representable and preserved.
        let max = i64::try_from(MAX_ASN1_TIMESTAMP_SECS).expect("fits in i64");
        assert_eq!(
            asn1_time_to_system_time(asn1(max)),
            UNIX_EPOCH + Duration::from_secs(MAX_ASN1_TIMESTAMP_SECS)
        );
    }

    #[test]
    fn host_semaphore_evicts_idle_at_cap() {
        let mut map = HashMap::new();
        for i in 0..4 {
            // Dropped immediately: only the map holds each semaphore (idle).
            drop(
                acquire_host_semaphore(&mut map, &format!("idle-{i}.example"), 4)
                    .expect("under cap"),
            );
        }
        assert_eq!(map.len(), 4);

        // At the cap, a NEW host must succeed by evicting idle entries —
        // the cap error is not sticky.
        let sem = acquire_host_semaphore(&mut map, "new-host.example", 4)
            .expect("idle eviction frees space for a new host");
        assert!(map.contains_key("new-host.example"));
        drop(sem);
    }

    #[test]
    fn host_semaphore_keeps_inflight_at_cap() {
        let mut map = HashMap::new();
        // Held across the cap check: simulates an in-flight fetch.
        let inflight = acquire_host_semaphore(&mut map, "busy.example", 3).expect("under cap");
        for i in 0..2 {
            drop(
                acquire_host_semaphore(&mut map, &format!("idle-{i}.example"), 3)
                    .expect("under cap"),
            );
        }
        assert_eq!(map.len(), 3);

        drop(
            acquire_host_semaphore(&mut map, "new-host.example", 3)
                .expect("idle entries evicted while in-flight survives"),
        );
        assert!(
            map.contains_key("busy.example"),
            "in-flight host must survive eviction"
        );
        assert!(map.contains_key("new-host.example"));
        drop(inflight);
    }

    #[test]
    fn host_semaphore_cap_error_when_all_inflight() {
        let mut map = HashMap::new();
        let held: Vec<_> = (0..2)
            .map(|i| {
                acquire_host_semaphore(&mut map, &format!("busy-{i}.example"), 2)
                    .expect("under cap")
            })
            .collect();

        let result = acquire_host_semaphore(&mut map, "new-host.example", 2);
        assert!(
            result.is_err(),
            "cap must still reject when every entry has an in-flight fetch"
        );
        drop(held);
    }
}