Skip to main content

rmcp_server_kit/
mtls_revocation.rs

1//! CDP-driven CRL revocation support for mTLS.
2//!
3//! When mTLS is configured with CRL checks enabled, startup performs a bounded
4//! bootstrap pass over the configured CA bundle, extracts CRL Distribution
5//! Point (CDP) URLs, fetches reachable CRLs, and builds the initial inner
6//! `rustls` verifier from that cache.
7//!
8//! During handshakes, the outer verifier remains stable for the lifetime of the
9//! TLS acceptor while its inner `WebPkiClientVerifier` is swapped atomically via
10//! `ArcSwap` as CRLs are discovered or refreshed. Discovery from connecting
11//! client certificates is fire-and-forget and never blocks the synchronous
12//! handshake path.
13//!
14//! Security note: CDP URLs are extracted from attacker-controllable client
15//! certs *before* chain validation. This is safe by design; see the
16//! `// SECURITY:` comment on `DynamicClientCertVerifier::verify_client_cert`
17//! for the full rationale before changing the discovery ordering.
18//!
19//! Semantics:
20//! - `crl_deny_on_unavailable = true` (default) => fail closed when a
21//!   certificate advertises CDP URLs whose revocation status is not yet
22//!   available. Denial requires *every* relevant CDP to be uncached, per
23//!   RFC 5280 6.3; denying on a single unavailable mirror would let an
24//!   attacker who blocks one CDP deny service.
25//! - `crl_deny_on_unavailable = false` => fail open with warn logs. Restores
26//!   the pre-3.9 behaviour and is strongly discouraged: a revoked
27//!   certificate is accepted whenever its CRL is unreachable.
28
29use std::{
30    collections::{HashMap, HashSet},
31    num::NonZeroU32,
32    pin::Pin,
33    sync::{Arc, Mutex},
34    time::{Duration, SystemTime, UNIX_EPOCH},
35};
36
37use arc_swap::ArcSwap;
38use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
39use rustls::{
40    DigitallySignedStruct, DistinguishedName, Error as TlsError, RootCertStore, SignatureScheme,
41    client::danger::HandshakeSignatureValid,
42    pki_types::{CertificateDer, CertificateRevocationListDer, UnixTime},
43    server::{
44        WebPkiClientVerifier,
45        danger::{ClientCertVerified, ClientCertVerifier},
46    },
47};
48use tokio::{
49    net::lookup_host,
50    sync::{RwLock, Semaphore, mpsc},
51    task::JoinSet,
52    time::{Instant, Sleep},
53};
54use tokio_util::sync::CancellationToken;
55use url::Url;
56use x509_parser::{
57    extensions::{DistributionPointName, GeneralName, ParsedExtension},
58    prelude::{FromDer, X509Certificate},
59    revocation_list::CertificateRevocationList,
60};
61
62use crate::{
63    auth::MtlsConfig,
64    error::RmcpServerKitError,
65    ssrf::{check_scheme, ip_block_reason, sanitized_url_for_log},
66};
67
68const BOOTSTRAP_TIMEOUT: Duration = Duration::from_secs(10);
69const MIN_AUTO_REFRESH: Duration = Duration::from_mins(10);
70const MAX_AUTO_REFRESH: Duration = Duration::from_hours(24);
71/// Connection timeout for CRL HTTP fetches. Independent of overall fetch
72/// timeout to bound time spent on unreachable hosts.
73const CRL_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
74/// Most distinct CDP URLs a single handshake will consider.
75///
76/// RFC 5280 4.2.1.13 treats multiple URIs inside one `DistributionPoint` as
77/// mirrors of the same CRL, so a conforming certificate needs only a handful.
78/// 64 sits far above any plausible legitimate certificate while bounding the
79/// work a peer can demand on the unauthenticated handshake path. Deliberately
80/// a private constant rather than a config field: no operator should need to
81/// tune it, and widening the public config surface for it would be worse.
82const MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE: usize = 64;
83
84/// Parsed CRL cached in memory and keyed by its source URL.
85#[derive(Clone, Debug)]
86#[non_exhaustive]
87pub struct CachedCrl {
88    /// DER bytes for the CRL.
89    pub der: CertificateRevocationListDer<'static>,
90    /// `thisUpdate` field from the CRL.
91    pub this_update: SystemTime,
92    /// `nextUpdate` field from the CRL, if present.
93    pub next_update: Option<SystemTime>,
94    /// Time the server fetched this CRL.
95    pub fetched_at: SystemTime,
96    /// Source URL used for retrieval.
97    pub source_url: String,
98}
99
100/// One atomically-published verifier generation.
101///
102/// SECURITY: `verifier`, `cached_urls`, and `committed_identities` are
103/// published together as a single immutable snapshot behind one [`ArcSwap`].
104/// Publishing them separately would let a handshake observe the coverage hint
105/// from one commit against the verifier of another — a mixed-generation read
106/// the fail-closed precheck cannot distinguish from out-of-band mutation.
107/// Readers load this state exactly once per handshake and both pre-check and
108/// enforce against it.
109pub(crate) struct VerifierState {
110    /// What rustls actually enforces for this generation.
111    verifier: Arc<dyn ClientCertVerifier>,
112    /// URLs whose CRL this generation's `verifier` genuinely enforces.
113    cached_urls: HashSet<String>,
114    /// Constant-cost identity of every committed [`CachedCrl`], keyed by URL.
115    ///
116    /// The precheck compares the live public `cache` against this index, so an
117    /// out-of-band write through the `pub` cache field is detected and denied
118    /// instead of silently claiming coverage the verifier does not provide.
119    committed_identities: HashMap<String, EntryIdentity>,
120}
121
122impl std::fmt::Debug for VerifierState {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        f.debug_struct("VerifierState")
125            .field("cached_urls_len", &self.cached_urls.len())
126            .field("committed_identities_len", &self.committed_identities.len())
127            .finish_non_exhaustive()
128    }
129}
130
131/// Constant-cost fingerprint of a cached CRL entry, recorded at commit time.
132///
133/// # What this does and does not guarantee
134///
135/// This detects **out-of-band mutation of the public [`CrlSet::cache`] field by
136/// non-adversarial code** — the API-misuse hazard that field creates. It is
137/// deliberately **not** a cryptographic integrity check and must never be
138/// described as one.
139///
140/// Hashing the DER instead would be a stronger check, but it was measured
141/// (`benches/crl_precheck.rs`) at 56 ms p95 per relevant cached CRL at the
142/// 5 MiB `crl_max_response_bytes` default, scaling linearly with an
143/// attacker-chosen CDP-URL count on the *unauthenticated* handshake path. That
144/// trades a local misuse tripwire for a remote CPU-amplification vulnerability,
145/// so identity comparison is used instead and the DER is never rescanned.
146///
147/// Known limit: a same-process caller that deliberately drops an entry and
148/// reallocates a replacement at the same address, with the same length and the
149/// same 32-byte head and tail, would not be detected. Such a caller already
150/// holds the cache write lock and is inside the trust boundary. The real fix is
151/// making `cache` private, which is a 4.0 change.
152#[derive(Clone, PartialEq, Eq)]
153struct EntryIdentity {
154    der_ptr: usize,
155    der_len: usize,
156    head: [u8; 32],
157    tail: [u8; 32],
158    this_update: SystemTime,
159    next_update: Option<SystemTime>,
160    fetched_at: SystemTime,
161    source_url: String,
162}
163
164/// Fingerprint one cache entry. Reads at most 64 bytes of DER regardless of
165/// CRL size, so handshake cost does not scale with `crl_max_response_bytes`.
166fn entry_identity(entry: &CachedCrl) -> EntryIdentity {
167    // Exhaustive destructuring is load-bearing: adding a field to `CachedCrl`
168    // must fail compilation here so someone decides whether it belongs in the
169    // identity, rather than silently falling outside mutation detection.
170    let CachedCrl {
171        der,
172        this_update,
173        next_update,
174        fetched_at,
175        source_url,
176    } = entry;
177
178    let bytes = der.as_ref();
179    let mut head = [0u8; 32];
180    let mut tail = [0u8; 32];
181    let sample = bytes.len().min(32);
182    if let Some(source) = bytes.get(..sample)
183        && let Some(target) = head.get_mut(..sample)
184    {
185        target.copy_from_slice(source);
186    }
187    if let Some(source) = bytes.get(bytes.len().saturating_sub(sample)..)
188        && let Some(target) = tail.get_mut(..sample)
189    {
190        target.copy_from_slice(source);
191    }
192
193    EntryIdentity {
194        der_ptr: bytes.as_ptr().addr(),
195        der_len: bytes.len(),
196        head,
197        tail,
198        this_update: *this_update,
199        next_update: *next_update,
200        fetched_at: *fetched_at,
201        source_url: source_url.clone(),
202    }
203}
204
205/// Fingerprint every entry of a cache map.
206fn crl_cache_identities<S: std::hash::BuildHasher>(
207    cache: &HashMap<String, CachedCrl, S>,
208) -> HashMap<String, EntryIdentity> {
209    cache
210        .iter()
211        .map(|(url, entry)| (url.clone(), entry_identity(entry)))
212        .collect()
213}
214
215/// Confirm the live cache still matches what `state` committed, for the CDP
216/// URLs this handshake would rely on.
217///
218/// SECURITY: the scope is `relevant_urls ∩ state.cached_urls` — the only URLs
219/// whose coverage claim can admit the handshake. A relevant URL missing from
220/// the live cache, missing an identity, or whose identity differs has had its
221/// entry mutated out of band: coverage is claimed but unproven.
222fn cache_matches_committed_identities(
223    cache: &HashMap<String, CachedCrl>,
224    state: &VerifierState,
225    relevant_urls: &[String],
226) -> bool {
227    relevant_urls
228        .iter()
229        .filter(|url| state.cached_urls.contains(*url))
230        .all(
231            |url| match (cache.get(url), state.committed_identities.get(url)) {
232                (Some(entry), Some(expected)) => entry_identity(entry) == *expected,
233                _ => false,
234            },
235        )
236}
237
238/// Shared CRL state backing the dynamic mTLS verifier.
239#[allow(
240    missing_debug_implementations,
241    reason = "contains ArcSwap and dyn verifier internals"
242)]
243#[non_exhaustive]
244pub struct CrlSet {
245    /// Single atomically-published generation of verifier + coverage hint +
246    /// digest index. See [`VerifierState`].
247    verifier_state: ArcSwap<VerifierState>,
248    /// Serializes commits end to end.
249    ///
250    /// SECURITY: the cache write lock is deliberately NOT the commit
251    /// transaction lock — holding it across `rebuild_verifier` would make the
252    /// verifier path's non-blocking `try_read` fail under ordinary refresh
253    /// load, turning a legitimate refresh into a spurious handshake denial.
254    /// Without this mutex, two commits could each snapshot the same old cache
255    /// and publish states whose `cached_urls` omit the other's URL, which is
256    /// exactly the desynchronisation the digest index exists to prevent.
257    commit_lock: tokio::sync::Mutex<()>,
258    /// Cached CRLs keyed by URL.
259    ///
260    /// # ⚠️ Deprecated
261    ///
262    /// Writing through this field bypasses the atomic commit path and
263    /// desynchronises the published coverage hint from the live verifier.
264    /// Since 3.9 such a write is **detected and fails the handshake closed**
265    /// rather than silently admitting a certificate whose revocation status
266    /// cannot be enforced. Reads remain safe but are not part of the supported
267    /// surface. The field becomes private in 4.0.
268    #[deprecated(
269        since = "3.9.0",
270        note = "mutating the CRL cache out of band is detected and denies handshakes; this field becomes private in 4.0"
271    )]
272    pub cache: RwLock<HashMap<String, CachedCrl>>,
273    /// Immutable client-auth root store.
274    pub roots: Arc<RootCertStore>,
275    /// mTLS CRL configuration.
276    pub config: MtlsConfig,
277    /// Fire-and-forget discovery channel for newly-seen CDP URLs.
278    pub discover_tx: mpsc::UnboundedSender<String>,
279    client: reqwest::Client,
280    /// URLs whose CRL is confirmed present in `cache`. Permanent dedup: a URL
281    /// here is never re-enqueued for discovery.
282    seen_urls: Mutex<HashSet<String>>,
283    /// URLs admitted to the discovery channel but not yet confirmed cached.
284    ///
285    /// This exists so a queued URL is not re-enqueued while its fetch is in
286    /// flight, WITHOUT permanently suppressing it. Promotion to `seen_urls`
287    /// happens only once the CRL is actually in the cache; a fetch error or a
288    /// cache-cap rejection clears the entry so a later handshake can retry.
289    /// Merging the two states is exactly the bug this separation fixes: a
290    /// first-fetch failure would otherwise suppress the URL for the process
291    /// lifetime, silently disabling revocation for that CDP.
292    pending_urls: Mutex<HashSet<String>>,
293    /// Global cap on simultaneous CRL HTTP fetches (SSRF amplification guard).
294    global_fetch_sem: Arc<Semaphore>,
295    /// Per-host serializer (one in-flight fetch per origin host). Bounded
296    /// by `crl_max_host_semaphores`; at the cap, idle entries are evicted
297    /// on demand (see [`acquire_host_semaphore`]), so the cap only rejects
298    /// genuinely concurrent fetch floods and is never a permanent lockout.
299    host_semaphores: Arc<tokio::sync::Mutex<HashMap<String, Arc<Semaphore>>>>,
300    /// Global rate-limiter on discovery URL submissions; protects against
301    /// cert-driven URL flooding by a malicious mTLS peer.
302    ///
303    /// Note: this ships as a process-global limiter; per-source-IP scoping
304    /// is deferred to a future release because the rustls
305    /// `verify_client_cert` callback does not carry a `SocketAddr` for the
306    /// peer. This is a CRL-discovery limiter in the TLS verifier path —
307    /// distinct from the bearer pre-auth limiter (`AuthState`), which is
308    /// already keyed per-IP via a bounded keyed governor and lives in the
309    /// ordinary request middleware path.
310    discovery_limiter: Arc<DefaultDirectRateLimiter>,
311    /// Cached cap on per-fetch response body size; copied from `config` so the
312    /// hot path doesn't re-read the (rarely changing) config struct.
313    max_response_bytes: u64,
314    last_cap_warn: Mutex<HashMap<&'static str, Instant>>,
315    /// Test-only hook fired immediately before a discovered URL becomes
316    /// observable on `discover_tx`.
317    ///
318    /// Exists because the "pending marker is inserted before the URL is
319    /// published" invariant cannot otherwise be observed deterministically:
320    /// the previous test raced a spinning thread against the scheduler and
321    /// could pass without ever exercising the interleaving.
322    #[cfg(any(test, feature = "test-helpers"))]
323    discovery_send_probe: Mutex<Option<DiscoverySendProbe>>,
324}
325
326/// Callback fired immediately before a discovered URL is published.
327#[cfg(any(test, feature = "test-helpers"))]
328type DiscoverySendProbe = Arc<dyn Fn(&CrlSet, &str) + Send + Sync>;
329
330impl CrlSet {
331    fn new(
332        roots: Arc<RootCertStore>,
333        config: MtlsConfig,
334        discover_tx: mpsc::UnboundedSender<String>,
335        initial_cache: HashMap<String, CachedCrl>,
336    ) -> Result<Arc<Self>, RmcpServerKitError> {
337        // M-H2: install the SSRF screening resolver on the CRL fetcher.
338        // CRL CDP URLs come from attacker-controllable client certs and
339        // their hosts are re-resolved per fetch -- exactly the TOCTOU
340        // class M-H2 closes. The allowlist is empty (default-strict),
341        // matching the existing CRL pre-flight posture; operators who
342        // need internal CDPs would extend this with the same
343        // CompiledSsrfAllowlist plumbing used by oauth.
344        let allowlist = Arc::new(crate::ssrf::CompiledSsrfAllowlist::default());
345        let resolver: Arc<dyn reqwest::dns::Resolve> =
346            Arc::new(crate::ssrf_resolver::SsrfScreeningResolver::new(
347                Arc::clone(&allowlist),
348                #[cfg(any(test, feature = "test-helpers"))]
349                Arc::new(std::sync::atomic::AtomicBool::new(false)),
350                #[cfg(not(any(test, feature = "test-helpers")))]
351                (),
352            ));
353
354        let client = reqwest::Client::builder()
355            // M-H2/N1: see oauth.rs::OauthHttpClient::build for rationale.
356            .no_proxy()
357            .dns_resolver(Arc::clone(&resolver))
358            .timeout(config.crl_fetch_timeout)
359            .connect_timeout(CRL_CONNECT_TIMEOUT)
360            .tcp_keepalive(None)
361            .redirect(reqwest::redirect::Policy::none())
362            .user_agent(format!("rmcp-server-kit/{}", env!("CARGO_PKG_VERSION")))
363            .build()
364            .map_err(|error| {
365                RmcpServerKitError::Startup(format!("CRL HTTP client init: {error}"))
366            })?;
367
368        let initial_verifier = rebuild_verifier(&roots, &config, &initial_cache)?;
369        let seen_urls = initial_cache.keys().cloned().collect::<HashSet<_>>();
370        // SECURITY: identities MUST be seeded here, not only on commit. `new`
371        // receives the bootstrap-fetched cache, so an index populated only by
372        // `commit_cache_update_atomically` would read every bootstrapped CRL
373        // as mutated out of band and fail mTLS closed at startup.
374        let initial_state = VerifierState {
375            verifier: initial_verifier,
376            cached_urls: seen_urls.clone(),
377            committed_identities: crl_cache_identities(&initial_cache),
378        };
379
380        // Defense in depth: normal server startup reaches this only through
381        // `Validated<McpServerConfig>`, but the public `bootstrap_fetch` helper
382        // and test-helper constructors accept a raw `MtlsConfig` directly.
383        let concurrency = config.crl_max_concurrent_fetches.max(1);
384        let global_fetch_sem = Arc::new(Semaphore::new(concurrency));
385        let host_semaphores = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
386
387        // Same raw-`MtlsConfig` bypass as above; keep a one-token minimum even
388        // when callers skip the startup validator.
389        let rate =
390            NonZeroU32::new(config.crl_discovery_rate_per_min.max(1)).unwrap_or(NonZeroU32::MIN);
391        let discovery_limiter = Arc::new(RateLimiter::direct(Quota::per_minute(rate)));
392
393        let max_response_bytes = config.crl_max_response_bytes;
394
395        #[allow(
396            deprecated,
397            reason = "constructing the struct necessarily names the deprecated field; the deprecation targets downstream mutation, not construction"
398        )]
399        Ok(Arc::new(Self {
400            verifier_state: ArcSwap::from_pointee(initial_state),
401            commit_lock: tokio::sync::Mutex::new(()),
402            cache: RwLock::new(initial_cache),
403            roots,
404            config,
405            discover_tx,
406            client,
407            seen_urls: Mutex::new(seen_urls),
408            pending_urls: Mutex::new(HashSet::new()),
409            global_fetch_sem,
410            host_semaphores,
411            discovery_limiter,
412            max_response_bytes,
413            last_cap_warn: Mutex::new(HashMap::new()),
414            #[cfg(any(test, feature = "test-helpers"))]
415            discovery_send_probe: Mutex::new(None),
416        }))
417    }
418
419    /// Fire the test-only pre-publication probe, if one is installed.
420    ///
421    /// The `Arc` is cloned out and the lock released before the callback runs,
422    /// so a probe may re-enter `CrlSet` state (`pending_urls`, `seen_urls`)
423    /// without deadlocking against this non-reentrant `std::sync::Mutex`.
424    #[cfg(any(test, feature = "test-helpers"))]
425    fn fire_discovery_send_probe(&self, url: &str) {
426        let probe = self
427            .discovery_send_probe
428            .lock()
429            .unwrap_or_else(std::sync::PoisonError::into_inner)
430            .clone();
431        if let Some(probe) = probe {
432            probe(self, url);
433        }
434    }
435
436    #[cfg(not(any(test, feature = "test-helpers")))]
437    #[inline]
438    #[allow(
439        clippy::unused_self,
440        reason = "the receiver keeps the call site identical across cfgs; production builds compile this to nothing"
441    )]
442    fn fire_discovery_send_probe(&self, _url: &str) {}
443
444    /// Test-only: observe every URL at the instant before it is published to
445    /// the discovery channel.
446    #[cfg(any(test, feature = "test-helpers"))]
447    #[doc(hidden)]
448    pub fn __test_set_discovery_send_probe(&self, probe: DiscoverySendProbe) {
449        *self
450            .discovery_send_probe
451            .lock()
452            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(probe);
453    }
454
455    fn should_warn_throttled(&self, which: &'static str) -> bool {
456        let now = Instant::now();
457        let cooldown = Duration::from_mins(1);
458        let mut guard = self
459            .last_cap_warn
460            .lock()
461            .unwrap_or_else(std::sync::PoisonError::into_inner);
462        let should_emit = guard
463            .get(which)
464            .is_none_or(|last| now.saturating_duration_since(*last) >= cooldown);
465        if should_emit {
466            guard.insert(which, now);
467        }
468        should_emit
469    }
470
471    fn warn_cap_exceeded_throttled(&self, which: &'static str) {
472        if self.should_warn_throttled(which) {
473            tracing::warn!(which = which, "CRL map cap exceeded; dropping newest entry");
474        }
475    }
476
477    /// Report a rejected certificate that advertised more distinct CDP URLs
478    /// than [`MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE`]. Distinct from both the
479    /// unavailable-CRL and the out-of-band-mutation denials, because this one
480    /// applies in fail-open mode too and has no opt-out.
481    /// Single in-crate entry point to the deprecated public `cache` field.
482    ///
483    /// Routing every internal use through here keeps the deprecation honest for
484    /// downstream callers while confining the `allow` to one site instead of
485    /// scattering it across every read.
486    #[allow(
487        deprecated,
488        reason = "the deprecation targets downstream out-of-band mutation; in-crate reads and the atomic commit path are the supported users of this field"
489    )]
490    fn cache_lock(&self) -> &RwLock<HashMap<String, CachedCrl>> {
491        &self.cache
492    }
493
494    fn warn_cdp_cap_exceeded_throttled(&self, observed: usize) {
495        if self.should_warn_throttled("cdp_url_cap") {
496            tracing::warn!(
497                observed = observed,
498                cap = MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE,
499                "crl_cdp_url_cap_exceeded: client certificate advertises more distinct CDP URLs than the per-handshake cap; rejecting as malformed"
500            );
501        }
502    }
503
504    /// Report a fail-closed denial caused by out-of-band mutation of the public
505    /// `cache` field, distinct from the ordinary unavailable-CRL denial so
506    /// operators can tell a mutated entry from a missing mirror.
507    fn warn_cache_tamper_throttled(&self) {
508        if self.should_warn_throttled("cache_entry_mismatch") {
509            tracing::warn!(
510                "crl_cache_out_of_band_mutation: live CRL cache does not match the committed identity index; denying handshake"
511            );
512        }
513    }
514
515    // cancel-safe: `commit_lock` is held across awaits, but every mutation
516    // before publication builds a local candidate. The cache swap and the
517    // `verifier_state.store` publication are adjacent with NO await between
518    // them, so a cancelled commit leaves either the old or the new generation
519    // -- never a half-applied mix.
520    async fn commit_cache_update_atomically(
521        &self,
522        inserts: Vec<(String, CachedCrl)>,
523        removals: &[String],
524    ) -> Result<bool, RmcpServerKitError> {
525        // SECURITY (lock order, load-bearing): hold `commit_lock` across the
526        // whole transaction — snapshot, rebuild, publish. The cache write lock
527        // is taken only for the paired `*cache = candidate` +
528        // `verifier_state.store(..)` publication, so readers synchronising on
529        // that same `RwLock` can never observe a commit half-applied, while
530        // `rebuild_verifier` and digesting stay off-lock and out of the
531        // verifier path's way.
532        let _commit = self.commit_lock.lock().await;
533
534        let mut candidate = self.cache_lock().read().await.clone();
535        let mut admitted_urls = Vec::new();
536
537        // POLICY: at cap the NEWEST entry is rejected, never an existing
538        // one (no LRU). Under adversarial unique-CDP churn an LRU would
539        // let an attacker evict the legitimate warm set by spamming
540        // throwaway CDP URLs; rejecting newcomers instead preserves
541        // revocation coverage for the established CA estate. Confirmed
542        // by Oracle review of the 1.13.0 rust-review fix plan.
543        for (url, cached) in inserts {
544            if candidate.len() >= self.config.crl_max_cache_entries && !candidate.contains_key(&url)
545            {
546                self.warn_cap_exceeded_throttled("cache");
547                continue;
548            }
549            candidate.insert(url.clone(), cached);
550            admitted_urls.push(url);
551        }
552
553        for url in removals {
554            candidate.remove(url);
555        }
556
557        let verifier = rebuild_verifier(&self.roots, &self.config, &candidate)?;
558
559        // SECURITY: identities are recomputed for EVERY entry, not only for
560        // what this commit wrote. `candidate` is a clone of the live cache and
561        // cloning a `CachedCrl` reallocates its DER, so every carried-forward
562        // entry has a new address. Carrying identities forward would make an
563        // unrelated commit invalidate every other URL and deny every later
564        // handshake. Recomputation is O(entries) pointer and scalar reads with
565        // no hashing, so there is nothing to gain by being incremental.
566        let new_state = Arc::new(VerifierState {
567            verifier,
568            cached_urls: candidate.keys().cloned().collect(),
569            committed_identities: crl_cache_identities(&candidate),
570        });
571        let changed = !admitted_urls.is_empty() || !removals.is_empty();
572
573        {
574            let mut cache = self.cache_lock().write().await;
575            let superseded = std::mem::replace(&mut *cache, candidate);
576            self.verifier_state.store(new_state);
577            drop(cache);
578            // Free the superseded map only AFTER releasing the write lock.
579            // Dropping it in place would deallocate one `String` and one DER
580            // buffer per cached CRL inside the publication window, which is
581            // precisely the window the verifier path's non-blocking `try_read`
582            // has to get through. Measured at 256 entries: ~15% of reader
583            // attempts blocked with the in-place drop, under 1% without it.
584            drop(superseded);
585        }
586
587        // A removed CRL must become fully re-discoverable, so clear the URL
588        // from BOTH dedup states. Clearing only `seen_urls` would leave a
589        // stale `pending_urls` entry suppressing re-enqueue forever.
590        {
591            let mut seen = self
592                .seen_urls
593                .lock()
594                .unwrap_or_else(std::sync::PoisonError::into_inner);
595            for url in removals {
596                seen.remove(url);
597            }
598        }
599        {
600            let mut pending = self
601                .pending_urls
602                .lock()
603                .unwrap_or_else(std::sync::PoisonError::into_inner);
604            for url in removals {
605                pending.remove(url);
606            }
607        }
608
609        Ok(changed)
610    }
611
612    /// Force an immediate refresh of all currently known CRL URLs.
613    ///
614    /// # Errors
615    ///
616    /// Returns an error if rebuilding the inner verifier fails.
617    pub async fn force_refresh(&self) -> Result<(), RmcpServerKitError> {
618        let urls = {
619            let cache = self.cache_lock().read().await;
620            cache.keys().cloned().collect::<Vec<_>>()
621        };
622        self.refresh_urls(urls).await
623    }
624
625    // cancel-safe: selects due URLs and delegates to 
efresh_urls, which
626    // stages results locally before a single atomic commit.
627    async fn refresh_due_urls(&self) -> Result<(), RmcpServerKitError> {
628        let now = SystemTime::now();
629        let urls = {
630            let cache = self.cache_lock().read().await;
631            cache
632                .iter()
633                .filter(|(_, cached)| {
634                    should_refresh_cached(cached, now, self.config.crl_refresh_interval)
635                })
636                .map(|(url, _)| url.clone())
637                .collect::<Vec<_>>()
638        };
639
640        if urls.is_empty() {
641            return Ok(());
642        }
643
644        self.refresh_urls(urls).await
645    }
646
647    // cancel-safe: fetch results accumulate in local insert/remove vectors and
648    // are applied only by commit_cache_update_atomically. Cancelling before
649    // that commit discards the batch and leaves the cache untouched.
650    async fn refresh_urls(&self, urls: Vec<String>) -> Result<(), RmcpServerKitError> {
651        let results = self.fetch_url_results(urls).await;
652        let now = SystemTime::now();
653        let cache = self.cache_lock().read().await;
654        let mut inserts = Vec::new();
655        let mut removals = Vec::new();
656
657        for (url, result) in results {
658            match result {
659                Ok(cached) => {
660                    inserts.push((url, cached));
661                }
662                Err(error) => {
663                    let remove_entry = cache.get(&url).is_some_and(|existing| {
664                        existing
665                            .next_update
666                            .and_then(|next| next.checked_add(self.config.crl_stale_grace))
667                            .is_some_and(|deadline| now > deadline)
668                    });
669                    tracing::warn!(url = %url, error = %error, "CRL refresh failed");
670                    if remove_entry {
671                        removals.push(url);
672                    }
673                }
674            }
675        }
676        drop(cache);
677
678        if !inserts.is_empty() || !removals.is_empty() {
679            let _ = self
680                .commit_cache_update_atomically(inserts, &removals)
681                .await?;
682        }
683
684        Ok(())
685    }
686
687    /// Fetch a CRL and commit it to the cache.
688    ///
689    /// Returns whether the CRL is actually present in the cache afterwards.
690    /// A successful HTTP fetch is NOT sufficient:
691    /// [`Self::commit_cache_update_atomically`] rejects new entries once
692    /// `crl_max_cache_entries` is reached. Only a URL that genuinely landed in
693    /// the cache may be promoted to the permanent `seen_urls` dedup set —
694    /// promoting on fetch success alone would suppress a URL that was never
695    /// cached, which is the same revocation-bypass this state split fixes.
696    // cancel-safe: commits only after gated_fetch returns, so cancellation
697    // during the fetch cannot promote a URL into the seen_urls dedup set.
698    async fn fetch_and_store_url(&self, url: String) -> Result<bool, RmcpServerKitError> {
699        let cached = gated_fetch(
700            &self.client,
701            &self.global_fetch_sem,
702            &self.host_semaphores,
703            &url,
704            self.config.crl_allow_http,
705            self.max_response_bytes,
706            self.config.crl_max_host_semaphores,
707        )
708        .await?;
709        let _ = self
710            .commit_cache_update_atomically(vec![(url.clone(), cached)], &[])
711            .await?;
712        Ok(self.cache_lock().read().await.contains_key(&url))
713    }
714
715    /// Promote a URL from the in-flight set to the permanent dedup set.
716    /// Called only once its CRL is confirmed present in the cache.
717    fn promote_pending_to_seen(&self, url: &str) {
718        {
719            let mut pending = self
720                .pending_urls
721                .lock()
722                .unwrap_or_else(std::sync::PoisonError::into_inner);
723            pending.remove(url);
724        }
725        let mut seen = self
726            .seen_urls
727            .lock()
728            .unwrap_or_else(std::sync::PoisonError::into_inner);
729        if seen.len() >= self.config.crl_max_seen_urls && !seen.contains(url) {
730            self.warn_cap_exceeded_throttled("seen_urls");
731            return;
732        }
733        seen.insert(url.to_owned());
734    }
735
736    /// Clear a URL's in-flight marker without promoting it, so a later
737    /// handshake can re-enqueue it. Used when the fetch failed or the cache
738    /// refused the entry.
739    fn clear_pending(&self, url: &str) {
740        let mut pending = self
741            .pending_urls
742            .lock()
743            .unwrap_or_else(std::sync::PoisonError::into_inner);
744        pending.remove(url);
745    }
746
747    /// Enqueue newly-seen CDP URLs and run the synchronous fail-closed
748    /// precheck.
749    ///
750    /// Returns `(deny, state)`. The caller MUST enforce with the returned
751    /// `state.verifier` rather than re-loading: pre-check and enforcement have
752    /// to observe the same verifier generation.
753    #[allow(
754        clippy::significant_drop_tightening,
755        reason = "the cache read guard is deliberately acquired BEFORE loading VerifierState and is released by the match that consumes it; tightening as the lint suggests would invert the lock order this precheck's generation-coherence depends on"
756    )]
757    fn note_discovered_urls(
758        &self,
759        end_entity_urls: &[String],
760        intermediate_urls: &[String],
761    ) -> (bool, Arc<VerifierState>) {
762        // INVARIANT: only called post-handshake from
763        // `DynamicClientCertVerifier::verify_client_cert`. The peer has
764        // already presented a chain that parses; this method must not panic
765        // under attacker-controlled URL contents.
766        //
767        // SECURITY: see `DynamicClientCertVerifier::verify_client_cert` for
768        // the rationale on why accepting URLs from an unverified cert is
769        // safe (no HTTP on this path; fetch is off-path and SSRF-gated).
770        let mut all_urls = Vec::with_capacity(end_entity_urls.len() + intermediate_urls.len());
771        all_urls.extend_from_slice(end_entity_urls);
772        all_urls.extend_from_slice(intermediate_urls);
773        all_urls.sort();
774        all_urls.dedup();
775
776        let relevant_urls = if self.config.crl_end_entity_only {
777            end_entity_urls
778        } else {
779            all_urls.as_slice()
780        };
781
782        // SECURITY: reject a certificate advertising an implausible number of
783        // distinct CDP URLs, BEFORE the discovery loop and BEFORE the
784        // fail-open branch. Every later step is linear in this peer-chosen
785        // count, so leaving it unbounded is the amplification primitive this
786        // cap exists to remove; and the sort/dedup/rate-limiter work is paid
787        // identically under `crl_deny_on_unavailable = false`, so capping only
788        // the fail-closed path would leave the amplifier fully intact.
789        //
790        // This is a MALFORMED-CERTIFICATE rejection, not a revocation-status
791        // denial: an operator who set `crl_deny_on_unavailable = false` opted
792        // out of unavailability denials, not out of this. It therefore has no
793        // opt-out and carries its own distinct log message.
794        if relevant_urls.len() > MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE {
795            self.warn_cdp_cap_exceeded_throttled(relevant_urls.len());
796            return (true, self.verifier_state.load_full());
797        }
798
799        // Snapshot both dedup sets under their locks; do NOT mutate yet.
800        // A URL is skipped if it is already cached (`seen_urls`) or already
801        // queued and awaiting its fetch (`pending_urls`). Promotion to
802        // `seen_urls` happens only after the CRL is confirmed in the cache,
803        // so a URL that loses the limiter race, hits a closed channel, fails
804        // to fetch, or is rejected by the cache cap stays retriable. Marking
805        // "seen" any earlier permanently black-holes the URL: every later
806        // handshake would treat it as known and skip discovery, while no CRL
807        // was ever cached. With `crl_deny_on_unavailable = true` that is a
808        // persistent handshake failure; with fail-open it silently disables
809        // revocation checking for that CDP for the process lifetime.
810        // SECURITY: discover only from `relevant_urls`, the same set the cap
811        // above governs. Reading `all_urls` here instead would let a peer with
812        // `crl_end_entity_only = true` drive discovery, pending-set growth, and
813        // limiter consumption from uncapped intermediate CDPs -- the exact
814        // amplification the cap exists to remove.
815        let candidates: Vec<String> = {
816            let seen = self
817                .seen_urls
818                .lock()
819                .unwrap_or_else(std::sync::PoisonError::into_inner);
820            let pending = self
821                .pending_urls
822                .lock()
823                .unwrap_or_else(std::sync::PoisonError::into_inner);
824            relevant_urls
825                .iter()
826                .filter(|url| !seen.contains(*url) && !pending.contains(*url))
827                .cloned()
828                .collect()
829        };
830
831        // Rate-limit gate: drop excess submissions on the floor with a WARN.
832        // The mTLS verifier must remain non-blocking, so we use the
833        // synchronous `check()` API and never await here.
834        for url in candidates {
835            if self.discovery_limiter.check().is_err() {
836                tracing::warn!(
837                    url = %url,
838                    "discovery_rate_limited: dropped CDP URL beyond per-minute cap (will be retried on next handshake observing this URL)"
839                );
840                continue;
841            }
842            let inserted = {
843                // Invariant: while a URL is observable by the refresher, its
844                // transient in-flight marker already exists, so every fetch
845                // settlement path can remove or promote the same marker.
846                let mut guard = self
847                    .pending_urls
848                    .lock()
849                    .unwrap_or_else(std::sync::PoisonError::into_inner);
850                if guard.contains(&url) {
851                    false
852                } else {
853                    if guard.len() >= self.config.crl_max_seen_urls {
854                        self.warn_cap_exceeded_throttled("pending_urls");
855                        break;
856                    }
857                    guard.insert(url.clone())
858                }
859            };
860            if !inserted {
861                continue;
862            }
863            self.fire_discovery_send_probe(&url);
864            if self.discover_tx.send(url.clone()).is_err() {
865                // Receiver gone (shutdown). Do NOT mark pending so the
866                // URL can be retried after a reload / restart.
867                self.clear_pending(&url);
868                tracing::debug!(
869                    url = %url,
870                    "discover channel closed; dropping CDP URL without marking pending"
871                );
872            }
873        }
874
875        if !self.config.crl_deny_on_unavailable {
876            return (false, self.verifier_state.load_full());
877        }
878
879        if relevant_urls.is_empty() {
880            return (false, self.verifier_state.load_full());
881        }
882
883        // SECURITY (lock order — must match `commit_cache_update_atomically`):
884        // take the cache read guard FIRST, then load the verifier state while
885        // still holding it. Commits publish `cache` and `verifier_state` under
886        // the same write lock, so this ordering makes a legitimate commit
887        // atomic to this reader; loading the state first could pair a new
888        // identity index with a pre-commit cache view and report a routine
889        // refresh as out-of-band mutation.
890        //
891        // `try_read` is mandatory: this runs inside the synchronous rustls
892        // verifier callback, where blocking and awaiting are forbidden.
893        let cache_guard = self.cache_lock().try_read();
894        let state = self.verifier_state.load_full();
895
896        // SECURITY: a failed `try_read` is NOT evidence of tampering, and must
897        // not deny on its own. `tokio::sync::RwLock` is write-preferring, so an
898        // ordinary refresh commit makes `try_read` fail — denying here would
899        // turn every legitimate CRL refresh into a self-inflicted handshake
900        // outage. It also buys no security: the handshake is enforced by the
901        // immutable `state.verifier` that was committed with `cached_urls`, not
902        // by the live map, so an unreadable cache simply means the public
903        // mirror could not be audited this time. Out-of-band mutation is still
904        // caught on every handshake that does get the lock.
905        if let Ok(cache) = cache_guard
906            && !cache_matches_committed_identities(&cache, &state, relevant_urls)
907        {
908            drop(cache);
909            self.warn_cache_tamper_throttled();
910            return (true, state);
911        }
912
913        // `all(..not cached..)` -- deny only when EVERY relevant CDP URL is
914        // uncached, not when any single one is. RFC 5280 4.2.1.13: "If the
915        // DistributionPointName contains multiple values, each name
916        // describes a different mechanism to obtain the same CRL." The
917        // URLs are therefore mirrors, and one successful fetch is
918        // sufficient revocation coverage; failing on a single unreachable
919        // mirror would let an attacker who can DoS one CDP host deny
920        // service to every client.
921        //
922        // Known limitation: multiple `DistributionPoint` *entries* (as
923        // opposed to multiple URIs inside one entry) may in principle be
924        // reason-partitioned scopes rather than mirrors, and this flattens
925        // them into a single URL set. That is safe against RFC-conforming
926        // issuers, because the same section requires "a conforming CA ...
927        // MUST include at least one DistributionPoint that points to a CRL
928        // that covers the certificate for all reasons", and the profile
929        // "RECOMMENDS against segmenting CRLs by reason code". Reason-code
930        // partitioning is not otherwise modelled here.
931        let deny = relevant_urls
932            .iter()
933            .all(|url| !state.cached_urls.contains(url));
934        (deny, state)
935    }
936
937    /// Test helper for constructing a CRL set from in-memory CRLs. Benign but
938    /// ungated public leak; test-only despite not requiring `test-helpers`.
939    ///
940    /// # Errors
941    ///
942    /// Returns an error if the verifier cannot be built from the provided CRLs.
943    #[doc(hidden)]
944    #[deprecated(
945        since = "3.9.0",
946        note = "test-only constructor that is ungated in 3.x by accident; it becomes feature-gated in 4.0"
947    )]
948    pub fn __test_with_prepopulated_crls(
949        roots: Arc<RootCertStore>,
950        config: MtlsConfig,
951        prefilled_crls: Vec<CertificateRevocationListDer<'static>>,
952    ) -> Result<Arc<Self>, RmcpServerKitError> {
953        let (discover_tx, discover_rx) = mpsc::unbounded_channel();
954        drop(discover_rx);
955
956        let mut initial_cache = HashMap::new();
957        for (index, der) in prefilled_crls.into_iter().enumerate() {
958            let source_url = format!("memory://crl/{index}");
959            let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;
960            initial_cache.insert(
961                source_url.clone(),
962                CachedCrl {
963                    der,
964                    this_update,
965                    next_update,
966                    fetched_at: SystemTime::now(),
967                    source_url,
968                },
969            );
970        }
971
972        Self::new(roots, config, discover_tx, initial_cache)
973    }
974
975    /// Test-only: same as [`Self::__test_with_prepopulated_crls`] but keeps and
976    /// returns the discover receiver. Benign but ungated public leak; test-only
977    /// despite not requiring `test-helpers`.
978    ///
979    /// # Errors
980    ///
981    /// Returns an error if the verifier cannot be built from the provided CRLs.
982    #[doc(hidden)]
983    #[deprecated(
984        since = "3.9.0",
985        note = "test-only constructor that is ungated in 3.x by accident; it becomes feature-gated in 4.0"
986    )]
987    pub fn __test_with_kept_receiver(
988        roots: Arc<RootCertStore>,
989        config: MtlsConfig,
990        prefilled_crls: Vec<CertificateRevocationListDer<'static>>,
991    ) -> Result<(Arc<Self>, mpsc::UnboundedReceiver<String>), RmcpServerKitError> {
992        let (discover_tx, discover_rx) = mpsc::unbounded_channel();
993
994        let mut initial_cache = HashMap::new();
995        for (index, der) in prefilled_crls.into_iter().enumerate() {
996            let source_url = format!("memory://crl/{index}");
997            let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;
998            initial_cache.insert(
999                source_url.clone(),
1000                CachedCrl {
1001                    der,
1002                    this_update,
1003                    next_update,
1004                    fetched_at: SystemTime::now(),
1005                    source_url,
1006                },
1007            );
1008        }
1009
1010        let crl_set = Self::new(roots, config, discover_tx, initial_cache)?;
1011        Ok((crl_set, discover_rx))
1012    }
1013
1014    /// # ⚠️ Security
1015    ///
1016    /// Availability hazard: bypasses discovery deduplication, consumes CDP
1017    /// discovery quota, and enqueues arbitrary URLs. Fetch-side SSRF, scheme,
1018    /// and concurrency caps still apply. Ungated public leak.
1019    #[doc(hidden)]
1020    pub fn __test_check_discovery_rate(&self, urls: &[String]) -> (usize, usize) {
1021        let mut accepted = 0usize;
1022        let mut dropped = 0usize;
1023        for url in urls {
1024            if self.discovery_limiter.check().is_ok() {
1025                let _ = self.discover_tx.send(url.clone());
1026                accepted += 1;
1027            } else {
1028                dropped += 1;
1029            }
1030        }
1031        (accepted, dropped)
1032    }
1033
1034    /// # ⚠️ Security
1035    ///
1036    /// State hazard: mutates discovery state; its closed-channel pending shim
1037    /// differs from production behaviour and can create state a real closed
1038    /// discovery channel would not record. Ungated public leak.
1039    #[doc(hidden)]
1040    pub fn __test_note_discovered_urls(&self, urls: &[String]) -> bool {
1041        let (missing_cached, _state) = self.note_discovered_urls(urls, &[]);
1042        if self.discover_tx.is_closed() {
1043            let already_seen: HashSet<String> = {
1044                let seen = self
1045                    .seen_urls
1046                    .lock()
1047                    .unwrap_or_else(std::sync::PoisonError::into_inner);
1048                urls.iter()
1049                    .filter(|url| seen.contains(*url))
1050                    .cloned()
1051                    .collect()
1052            };
1053            let mut pending = self
1054                .pending_urls
1055                .lock()
1056                .unwrap_or_else(std::sync::PoisonError::into_inner);
1057            for url in urls {
1058                if already_seen.contains(url) || pending.contains(url) {
1059                    continue;
1060                }
1061                if pending.len() >= self.config.crl_max_seen_urls {
1062                    self.warn_cap_exceeded_throttled("pending_urls");
1063                    break;
1064                }
1065                pending.insert(url.clone());
1066            }
1067        }
1068        missing_cached
1069    }
1070
1071    /// Test-only: invoke the real precheck with separate end-entity and
1072    /// intermediate CDP sets.
1073    #[cfg(any(test, feature = "test-helpers"))]
1074    #[doc(hidden)]
1075    pub fn __test_note_discovered_urls_by_cert(
1076        &self,
1077        end_entity_urls: &[String],
1078        intermediate_urls: &[String],
1079    ) -> bool {
1080        self.note_discovered_urls(end_entity_urls, intermediate_urls)
1081            .0
1082    }
1083
1084    /// Test-only: report whether a URL is suppressed from re-discovery. Benign
1085    /// inspection helper, but an ungated public leak available without
1086    /// `test-helpers`; use only in tests.
1087    #[doc(hidden)]
1088    pub fn __test_is_seen(&self, url: &str) -> bool {
1089        let in_seen = {
1090            let seen = self
1091                .seen_urls
1092                .lock()
1093                .unwrap_or_else(std::sync::PoisonError::into_inner);
1094            seen.contains(url)
1095        };
1096        if in_seen {
1097            return true;
1098        }
1099        let pending = self
1100            .pending_urls
1101            .lock()
1102            .unwrap_or_else(std::sync::PoisonError::into_inner);
1103        pending.contains(url)
1104    }
1105
1106    /// Test-only: report whether a URL reached the PERMANENT dedup set,
1107    /// which happens only after its CRL is confirmed present in the cache.
1108    /// A URL that was merely queued, or whose fetch failed, is not counted.
1109    #[cfg(any(test, feature = "test-helpers"))]
1110    #[doc(hidden)]
1111    pub fn __test_is_permanently_seen(&self, url: &str) -> bool {
1112        let seen = self
1113            .seen_urls
1114            .lock()
1115            .unwrap_or_else(std::sync::PoisonError::into_inner);
1116        seen.contains(url)
1117    }
1118
1119    /// # ⚠️ Security
1120    ///
1121    /// Caller-supplied `admitted = true` promotes to `seen_urls` without
1122    /// verifying the CRL is cached; a wrongly promoted URL is never re-enqueued
1123    /// for the process lifetime.
1124    #[cfg(any(test, feature = "test-helpers"))]
1125    #[doc(hidden)]
1126    pub fn __test_settle_pending(&self, url: &str, admitted: bool) {
1127        if admitted {
1128            self.promote_pending_to_seen(url);
1129        } else {
1130            self.clear_pending(url);
1131        }
1132    }
1133
1134    /// Test-only: current count of host semaphores. Used by
1135    /// `tests/crl_map_bounds.rs` to assert the cap is enforced.
1136    #[cfg(any(test, feature = "test-helpers"))]
1137    #[doc(hidden)]
1138    pub fn __test_host_semaphore_count(&self) -> usize {
1139        self.host_semaphores
1140            .try_lock()
1141            .map_or(0, |guard| guard.len())
1142    }
1143
1144    /// Test-only: current number of entries in the CRL cache.
1145    #[cfg(any(test, feature = "test-helpers"))]
1146    #[doc(hidden)]
1147    pub fn __test_cache_len(&self) -> usize {
1148        self.cache_lock().try_read().map_or(0, |guard| guard.len())
1149    }
1150
1151    /// Test-only: whether a specific URL is currently cached.
1152    #[cfg(any(test, feature = "test-helpers"))]
1153    #[doc(hidden)]
1154    pub fn __test_cache_contains(&self, url: &str) -> bool {
1155        self.cache_lock()
1156            .try_read()
1157            .is_ok_and(|guard| guard.contains_key(url))
1158    }
1159
1160    /// Test-only: whether a URL is advertised to the fail-closed precheck as
1161    /// present in the live verifier.
1162    #[cfg(any(test, feature = "test-helpers"))]
1163    #[doc(hidden)]
1164    pub fn __test_cached_url_contains(&self, url: &str) -> bool {
1165        self.verifier_state.load().cached_urls.contains(url)
1166    }
1167
1168    /// # ⚠️ Security
1169    ///
1170    /// Calls `gated_fetch` directly, bypassing `note_discovered_urls` and
1171    /// therefore `discovery_limiter.check()`, the per-minute CDP discovery cap.
1172    /// SSRF, scheme, and concurrency caps still apply.
1173    #[cfg(any(test, feature = "test-helpers"))]
1174    #[doc(hidden)]
1175    pub async fn __test_trigger_fetch(&self, url: &str) -> Result<(), RmcpServerKitError> {
1176        if let Err(error) = gated_fetch(
1177            &self.client,
1178            &self.global_fetch_sem,
1179            &self.host_semaphores,
1180            url,
1181            self.config.crl_allow_http,
1182            self.max_response_bytes,
1183            self.config.crl_max_host_semaphores,
1184        )
1185        .await
1186        {
1187            if error
1188                .to_string()
1189                .contains("crl_host_semaphore_cap_exceeded")
1190            {
1191                Err(error)
1192            } else {
1193                Ok(())
1194            }
1195        } else {
1196            Ok(())
1197        }
1198    }
1199
1200    /// Test-only: insert through `commit_cache_update_atomically`, preserving
1201    /// normal cache/verifier publication. Best-effort: discards verifier
1202    /// rebuild errors, so an invalid CRL is a silent no-op.
1203    #[cfg(any(test, feature = "test-helpers"))]
1204    #[doc(hidden)]
1205    pub async fn __test_insert_cache(&self, url: &str, cached: CachedCrl) {
1206        let _ = self
1207            .commit_cache_update_atomically(vec![(url.to_owned(), cached)], &[])
1208            .await;
1209    }
1210
1211    /// Test-only: direct cache insertion that returns verifier rebuild errors.
1212    #[cfg(any(test, feature = "test-helpers"))]
1213    #[doc(hidden)]
1214    pub async fn __test_try_insert_cache(
1215        &self,
1216        url: &str,
1217        cached: CachedCrl,
1218    ) -> Result<bool, RmcpServerKitError> {
1219        self.commit_cache_update_atomically(vec![(url.to_owned(), cached)], &[])
1220            .await
1221    }
1222
1223    /// # ⚠️ Security
1224    ///
1225    /// Writes into `cache` directly, bypassing `commit_cache_update_atomically`
1226    /// and publication ordering at lines 283-291. This can desynchronise
1227    /// `cached_urls` from `inner_verifier`; with default
1228    /// `crl_deny_on_unavailable = true`, the lines 547-579 precheck trusts
1229    /// `cached_urls` and can admit a certificate whose revocation status is
1230    /// unenforceable.
1231    #[cfg(any(test, feature = "test-helpers"))]
1232    #[doc(hidden)]
1233    pub async fn __test_replace_cache_entry_unverified(&self, url: &str, cached: CachedCrl) {
1234        let mut cache = self.cache_lock().write().await;
1235        cache.insert(url.to_owned(), cached);
1236    }
1237
1238    /// # ⚠️ Security
1239    ///
1240    /// Lets a caller-supplied URL reach `refresh_urls` and then `gated_fetch`,
1241    /// bypassing normal CDP discovery admission, deduplication, and rate-limit
1242    /// checks.
1243    #[cfg(any(test, feature = "test-helpers"))]
1244    #[doc(hidden)]
1245    pub async fn __test_trigger_refresh_url(&self, url: &str) -> Result<(), RmcpServerKitError> {
1246        self.refresh_urls(vec![url.to_owned()]).await
1247    }
1248
1249    // cancel-safe (cache integrity): `join_next` fills a local `Vec`; dropping
1250    // the `JoinSet` aborts unfinished `gated_fetch` before any cache commit.
1251    // A cancelled fetch leaves at most an idle bounded host semaphore.
1252    async fn fetch_url_results(
1253        &self,
1254        urls: Vec<String>,
1255    ) -> Vec<(String, Result<CachedCrl, RmcpServerKitError>)> {
1256        let mut tasks = JoinSet::new();
1257        for url in urls {
1258            let client = self.client.clone();
1259            let global_sem = Arc::clone(&self.global_fetch_sem);
1260            let host_map = Arc::clone(&self.host_semaphores);
1261            let allow_http = self.config.crl_allow_http;
1262            let max_bytes = self.max_response_bytes;
1263            let max_host_semaphores = self.config.crl_max_host_semaphores;
1264            tasks.spawn(async move {
1265                let result = gated_fetch(
1266                    &client,
1267                    &global_sem,
1268                    &host_map,
1269                    &url,
1270                    allow_http,
1271                    max_bytes,
1272                    max_host_semaphores,
1273                )
1274                .await;
1275                (url, result)
1276            });
1277        }
1278
1279        let mut results = Vec::new();
1280        while let Some(joined) = tasks.join_next().await {
1281            match joined {
1282                Ok(result) => results.push(result),
1283                Err(error) => {
1284                    tracing::warn!(error = %error, "CRL refresh task join failed");
1285                }
1286            }
1287        }
1288
1289        results
1290    }
1291}
1292
1293#[cfg(any(test, feature = "test-helpers"))]
1294const SYNTHETIC_TEST_CRL_DER: &[u8] = &[
1295    48, 129, 199, 48, 110, 2, 1, 1, 48, 10, 6, 8, 42, 134, 72, 206, 61, 4, 3, 2, 48, 14, 49, 12,
1296    48, 10, 6, 3, 85, 4, 3, 12, 3, 99, 114, 108, 23, 13, 50, 54, 48, 49, 48, 49, 48, 48, 48, 48,
1297    48, 48, 90, 23, 13, 50, 55, 48, 49, 48, 49, 48, 48, 48, 48, 48, 48, 90, 160, 47, 48, 45, 48,
1298    31, 6, 3, 85, 29, 35, 4, 24, 48, 22, 128, 20, 14, 62, 48, 146, 7, 182, 179, 215, 90, 226, 214,
1299    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,
1300    42, 134, 72, 206, 61, 4, 3, 2, 3, 73, 0, 48, 70, 2, 33, 0, 250, 240, 103, 87, 60, 78, 208, 171,
1301    184, 206, 117, 134, 236, 234, 53, 115, 122, 90, 64, 217, 146, 27, 32, 103, 170, 222, 240, 159,
1302    137, 187, 116, 6, 2, 33, 0, 188, 23, 204, 232, 130, 84, 135, 249, 43, 208, 224, 220, 202, 57,
1303    98, 140, 4, 251, 148, 189, 105, 68, 105, 40, 53, 180, 208, 38, 193, 120, 118, 100,
1304];
1305
1306impl CachedCrl {
1307    /// Test-only: synthesize a cache entry that looks valid, `next_update`
1308    /// = now + 24h. Fields used only to populate the HashMap — the bytes
1309    /// are a minimal CRL-shape that won't be parsed by tests.
1310    #[cfg(any(test, feature = "test-helpers"))]
1311    #[doc(hidden)]
1312    #[must_use]
1313    pub fn __test_synthetic(now: SystemTime) -> Self {
1314        Self {
1315            der: CertificateRevocationListDer::from(SYNTHETIC_TEST_CRL_DER.to_vec()),
1316            this_update: now,
1317            next_update: now.checked_add(Duration::from_hours(24)),
1318            fetched_at: now,
1319            source_url: "test://synthetic".to_owned(),
1320        }
1321    }
1322
1323    /// Test-only: synthesize a STALE cache entry (`next_update` in the
1324    /// deep past so `is_stale_beyond_grace` fires with any sensible
1325    /// `crl_stale_grace`).
1326    #[cfg(any(test, feature = "test-helpers"))]
1327    #[doc(hidden)]
1328    #[must_use]
1329    pub fn __test_stale(reference_past: SystemTime) -> Self {
1330        Self {
1331            der: CertificateRevocationListDer::from(vec![0x30, 0x00]),
1332            this_update: reference_past,
1333            next_update: Some(reference_past),
1334            fetched_at: reference_past,
1335            source_url: "test://stale".to_owned(),
1336        }
1337    }
1338}
1339
1340/// Stable outer verifier that delegates all TLS verification behavior to the
1341/// atomically swappable inner verifier.
1342pub struct DynamicClientCertVerifier {
1343    inner: Arc<CrlSet>,
1344    dn_subjects: Vec<DistinguishedName>,
1345}
1346
1347impl DynamicClientCertVerifier {
1348    /// Construct a new dynamic verifier from a shared [`CrlSet`].
1349    #[must_use]
1350    pub fn new(inner: Arc<CrlSet>) -> Self {
1351        Self {
1352            dn_subjects: inner.roots.subjects(),
1353            inner,
1354        }
1355    }
1356}
1357
1358impl std::fmt::Debug for DynamicClientCertVerifier {
1359    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1360        f.debug_struct("DynamicClientCertVerifier")
1361            .field("dn_subjects_len", &self.dn_subjects.len())
1362            .finish_non_exhaustive()
1363    }
1364}
1365
1366impl ClientCertVerifier for DynamicClientCertVerifier {
1367    fn offer_client_auth(&self) -> bool {
1368        let state = self.inner.verifier_state.load();
1369        state.verifier.offer_client_auth()
1370    }
1371
1372    fn client_auth_mandatory(&self) -> bool {
1373        let state = self.inner.verifier_state.load();
1374        state.verifier.client_auth_mandatory()
1375    }
1376
1377    fn root_hint_subjects(&self) -> &[DistinguishedName] {
1378        &self.dn_subjects
1379    }
1380
1381    fn verify_client_cert(
1382        &self,
1383        end_entity: &CertificateDer<'_>,
1384        intermediates: &[CertificateDer<'_>],
1385        now: UnixTime,
1386    ) -> Result<ClientCertVerified, TlsError> {
1387        // SECURITY: extracting CDP URLs from an unverified client cert
1388        // here is intentional. No HTTP happens on this path -- the call
1389        // to `note_discovered_urls` only enqueues onto a bounded,
1390        // rate-limited channel. The actual fetch runs off-path in
1391        // `run_crl_refresher` and is gated by SSRF screening
1392        // (`src/ssrf.rs`), body-size cap, deadline, and the
1393        // `crl_allow_http` policy. CRLs are CA-signed (RFC 5280 §5), so
1394        // http(s) CDP URLs are protocol design, not an SSRF sink. The
1395        // discovery must happen BEFORE delegating to the inner verifier
1396        // so `crl_deny_on_unavailable = true` can fail-closed on a
1397        // never-fetched CDP. Do NOT reorder.
1398        let mut end_entity_urls =
1399            extract_cdp_urls(end_entity.as_ref(), self.inner.config.crl_allow_http);
1400        end_entity_urls.sort();
1401        end_entity_urls.dedup();
1402
1403        let mut intermediate_urls = Vec::new();
1404        for intermediate in intermediates {
1405            intermediate_urls.extend(extract_cdp_urls(
1406                intermediate.as_ref(),
1407                self.inner.config.crl_allow_http,
1408            ));
1409        }
1410        intermediate_urls.sort();
1411        intermediate_urls.dedup();
1412
1413        let (revocation_unavailable, state) = self
1414            .inner
1415            .note_discovered_urls(&end_entity_urls, &intermediate_urls);
1416        if revocation_unavailable {
1417            return Err(TlsError::General(
1418                "client certificate revocation status unavailable".to_owned(),
1419            ));
1420        }
1421
1422        state
1423            .verifier
1424            .verify_client_cert(end_entity, intermediates, now)
1425    }
1426
1427    fn verify_tls12_signature(
1428        &self,
1429        message: &[u8],
1430        cert: &CertificateDer<'_>,
1431        dss: &DigitallySignedStruct,
1432    ) -> Result<HandshakeSignatureValid, TlsError> {
1433        let state = self.inner.verifier_state.load();
1434        state.verifier.verify_tls12_signature(message, cert, dss)
1435    }
1436
1437    fn verify_tls13_signature(
1438        &self,
1439        message: &[u8],
1440        cert: &CertificateDer<'_>,
1441        dss: &DigitallySignedStruct,
1442    ) -> Result<HandshakeSignatureValid, TlsError> {
1443        let state = self.inner.verifier_state.load();
1444        state.verifier.verify_tls13_signature(message, cert, dss)
1445    }
1446
1447    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
1448        let state = self.inner.verifier_state.load();
1449        state.verifier.supported_verify_schemes()
1450    }
1451
1452    fn requires_raw_public_keys(&self) -> bool {
1453        let state = self.inner.verifier_state.load();
1454        state.verifier.requires_raw_public_keys()
1455    }
1456}
1457
1458/// Extract CRL Distribution Point URLs from a DER-encoded certificate.
1459///
1460/// URLs are validated with `url::Url::parse` (case-insensitive scheme handling)
1461/// and filtered through an internal scheme guard. Malformed URLs, URLs
1462/// using disallowed schemes, and URLs carrying embedded credentials
1463/// (userinfo) are silently dropped. SSRF defenses against private
1464/// IP literals and metadata endpoints are applied later, at fetch time, after
1465/// DNS resolution.
1466#[must_use]
1467pub fn extract_cdp_urls(cert_der: &[u8], allow_http: bool) -> Vec<String> {
1468    let Ok((_, cert)) = X509Certificate::from_der(cert_der) else {
1469        return Vec::new();
1470    };
1471
1472    let mut urls = Vec::new();
1473    for ext in cert.extensions() {
1474        if let ParsedExtension::CRLDistributionPoints(cdps) = ext.parsed_extension() {
1475            for point in cdps.iter() {
1476                if let Some(DistributionPointName::FullName(names)) = &point.distribution_point {
1477                    for name in names {
1478                        if let GeneralName::URI(uri) = name {
1479                            let raw = *uri;
1480                            let Ok(parsed) = Url::parse(raw) else {
1481                                // `?raw` (Debug) escapes control characters the
1482                                // failed parse may have left in this
1483                                // attacker-supplied string.
1484                                tracing::debug!(url = ?raw, "CDP URL parse failed; dropped");
1485                                continue;
1486                            };
1487                            if let Err(reason) = check_scheme(&parsed, allow_http) {
1488                                tracing::debug!(
1489                                    url = %sanitized_url_for_log(&parsed),
1490                                    reason,
1491                                    "CDP URL rejected by scheme guard; dropped"
1492                                );
1493                                continue;
1494                            }
1495                            urls.push(parsed.into());
1496                        }
1497                    }
1498                }
1499            }
1500        }
1501    }
1502
1503    urls
1504}
1505
1506/// Bound the startup CDP fan-out to the same limit the steady-state cache obeys.
1507///
1508/// SECURITY: `bootstrap_fetch` is a public helper taking a raw [`MtlsConfig`]
1509/// and raw CA certificates, so it is reachable without
1510/// `McpServerConfig::validate`. A broad CA bundle would otherwise spawn one
1511/// fetch task and one cache entry per distinct CDP URL, unbounded by the cap
1512/// that governs every other cache write.
1513///
1514/// Extracted from `bootstrap_fetch` so the bound is testable at all: that
1515/// function cannot be driven from a test without weakening production SSRF
1516/// screening, which `bootstrap_cache_cap_is_applied_before_crl_set_publication`
1517/// documents at length and explicitly forbids.
1518fn cap_bootstrap_urls(urls: &mut Vec<String>, cap: usize) {
1519    if urls.len() > cap {
1520        tracing::warn!(
1521            discovered = urls.len(),
1522            cap,
1523            "CRL bootstrap: CA chain advertises more distinct CDP URLs than \
1524             crl_max_cache_entries; fetching only the first {cap} after dedup"
1525        );
1526        urls.truncate(cap);
1527    }
1528}
1529
1530/// Bootstrap the CRL cache by extracting CDP URLs from the CA chain and
1531/// fetching any reachable CRLs with a 10-second total deadline.
1532///
1533/// # Errors
1534///
1535/// Returns an error if the initial verifier cannot be built.
1536#[allow(
1537    clippy::cognitive_complexity,
1538    reason = "bootstrap coordinates timeout, parallel fetches, and partial-cache recovery"
1539)]
1540// cancel-safe: CRL cache state is local until final `CrlSet::new`; timeout or
1541// cancellation drops the `JoinSet`, aborting in-flight `gated_fetch` before
1542// publication. Bootstrap host semaphores are local and drop with this future.
1543pub async fn bootstrap_fetch(
1544    roots: Arc<RootCertStore>,
1545    ca_certs: &[CertificateDer<'static>],
1546    config: MtlsConfig,
1547) -> Result<(Arc<CrlSet>, mpsc::UnboundedReceiver<String>), RmcpServerKitError> {
1548    let (discover_tx, discover_rx) = mpsc::unbounded_channel();
1549
1550    let mut urls = ca_certs
1551        .iter()
1552        .flat_map(|cert| extract_cdp_urls(cert.as_ref(), config.crl_allow_http))
1553        .collect::<Vec<_>>();
1554    urls.sort();
1555    urls.dedup();
1556    cap_bootstrap_urls(&mut urls, config.crl_max_cache_entries);
1557
1558    // M-H2: same SSRF resolver hardening as CrlSet::new -- bootstrap
1559    // fetches the same attacker-controlled CDP URLs, just earlier in
1560    // the lifecycle.
1561    let bootstrap_allowlist = Arc::new(crate::ssrf::CompiledSsrfAllowlist::default());
1562    let bootstrap_resolver: Arc<dyn reqwest::dns::Resolve> =
1563        Arc::new(crate::ssrf_resolver::SsrfScreeningResolver::new(
1564            Arc::clone(&bootstrap_allowlist),
1565            #[cfg(any(test, feature = "test-helpers"))]
1566            Arc::new(std::sync::atomic::AtomicBool::new(false)),
1567            #[cfg(not(any(test, feature = "test-helpers")))]
1568            (),
1569        ));
1570
1571    let client = reqwest::Client::builder()
1572        // M-H2/N1: see oauth.rs::OauthHttpClient::build for rationale.
1573        .no_proxy()
1574        .dns_resolver(Arc::clone(&bootstrap_resolver))
1575        .timeout(config.crl_fetch_timeout)
1576        .connect_timeout(CRL_CONNECT_TIMEOUT)
1577        .tcp_keepalive(None)
1578        .redirect(reqwest::redirect::Policy::none())
1579        .user_agent(format!("rmcp-server-kit/{}", env!("CARGO_PKG_VERSION")))
1580        .build()
1581        .map_err(|error| RmcpServerKitError::Startup(format!("CRL HTTP client init: {error}")))?;
1582
1583    // Bootstrap shares the same global concurrency + per-host cap as the
1584    // hot-path verifier so a maliciously broad CA chain cannot overwhelm
1585    // the network at startup.
1586    // Defense in depth: this public helper accepts raw `MtlsConfig` directly,
1587    // so callers can bypass `McpServerConfig::validate`.
1588    let bootstrap_concurrency = config.crl_max_concurrent_fetches.max(1);
1589    let global_sem = Arc::new(Semaphore::new(bootstrap_concurrency));
1590    let host_semaphores = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
1591    let allow_http = config.crl_allow_http;
1592    let max_bytes = config.crl_max_response_bytes;
1593    let max_host_semaphores = config.crl_max_host_semaphores;
1594
1595    let mut initial_cache = HashMap::new();
1596    let mut tasks = JoinSet::new();
1597    for url in &urls {
1598        let client = client.clone();
1599        let url = url.clone();
1600        let global_sem = Arc::clone(&global_sem);
1601        let host_semaphores = Arc::clone(&host_semaphores);
1602        tasks.spawn(async move {
1603            let result = gated_fetch(
1604                &client,
1605                &global_sem,
1606                &host_semaphores,
1607                &url,
1608                allow_http,
1609                max_bytes,
1610                max_host_semaphores,
1611            )
1612            .await;
1613            (url, result)
1614        });
1615    }
1616
1617    let timeout: Sleep = tokio::time::sleep(BOOTSTRAP_TIMEOUT);
1618    tokio::pin!(timeout);
1619
1620    while !tasks.is_empty() {
1621        // cancel-safe: pinned Sleep and JoinSet::join_next are cancel-safe
1622        // (tokio docs); on timeout the loop breaks and dropping the JoinSet
1623        // aborts remaining fetches — the intended deadline behavior.
1624        tokio::select! {
1625            () = &mut timeout => {
1626                tracing::warn!("CRL bootstrap timed out after {:?}", BOOTSTRAP_TIMEOUT);
1627                break;
1628            }
1629            maybe_joined = tasks.join_next() => {
1630                let Some(joined) = maybe_joined else {
1631                    break;
1632                };
1633                match joined {
1634                    Ok((url, Ok(cached))) => {
1635                        initial_cache.insert(url, cached);
1636                    }
1637                    Ok((url, Err(error))) => {
1638                        tracing::warn!(url = %url, error = %error, "CRL bootstrap fetch failed");
1639                    }
1640                    Err(error) => {
1641                        tracing::warn!(error = %error, "CRL bootstrap task join failed");
1642                    }
1643                }
1644            }
1645        }
1646    }
1647
1648    let set = new_crl_set_from_bootstrap_cache(roots, config, discover_tx, initial_cache)?;
1649    Ok((set, discover_rx))
1650}
1651
1652fn new_crl_set_from_bootstrap_cache(
1653    roots: Arc<RootCertStore>,
1654    config: MtlsConfig,
1655    discover_tx: mpsc::UnboundedSender<String>,
1656    mut initial_cache: HashMap<String, CachedCrl>,
1657) -> Result<Arc<CrlSet>, RmcpServerKitError> {
1658    apply_bootstrap_cache_cap(&mut initial_cache, config.crl_max_cache_entries);
1659    CrlSet::new(roots, config, discover_tx, initial_cache)
1660}
1661
1662fn apply_bootstrap_cache_cap(
1663    initial_cache: &mut HashMap<String, CachedCrl>,
1664    max_cache_entries: usize,
1665) {
1666    if initial_cache.len() <= max_cache_entries {
1667        return;
1668    }
1669
1670    let mut urls = initial_cache.keys().cloned().collect::<Vec<_>>();
1671    urls.sort();
1672    for url in urls.into_iter().skip(max_cache_entries) {
1673        initial_cache.remove(&url);
1674    }
1675}
1676
1677/// Run the CRL refresher loop until shutdown.
1678#[allow(
1679    clippy::cognitive_complexity,
1680    reason = "refresher loop intentionally handles shutdown, timer, and discovery in one select"
1681)]
1682// cancel-safe, including under abort: cooperative `shutdown` breaks the loop at
1683// a settlement point, and the discovery arm holds a `PendingUrlGuard` across
1684// `fetch_and_store_url`, so a `JoinHandle::abort` that drops the future mid-await
1685// still clears the transient `pending_urls` marker via `Drop`. Without that
1686// guard a stale marker would suppress re-enqueue forever (see the note above
1687// `seen_urls`/`pending_urls` clearing) and silently narrow revocation coverage.
1688pub async fn run_crl_refresher(
1689    set: Arc<CrlSet>,
1690    mut discover_rx: mpsc::UnboundedReceiver<String>,
1691    shutdown: CancellationToken,
1692) {
1693    let mut refresh_sleep = schedule_next_refresh(&set).await;
1694
1695    loop {
1696        // cancel-safe: CancellationToken::cancelled, pinned &mut Sleep, and
1697        // mpsc::UnboundedReceiver::recv are all cancel-safe (tokio docs);
1698        // refresh work happens inside arm bodies, never in the raced futures.
1699        tokio::select! {
1700            () = shutdown.cancelled() => {
1701                break;
1702            }
1703            () = &mut refresh_sleep => {
1704                if let Err(error) = set.refresh_due_urls().await {
1705                    tracing::warn!(error = %error, "CRL periodic refresh failed");
1706                }
1707                refresh_sleep = schedule_next_refresh(&set).await;
1708            }
1709            maybe_url = discover_rx.recv() => {
1710                let Some(url) = maybe_url else {
1711                    break;
1712                };
1713                let pending_guard = PendingUrlGuard::armed(Arc::clone(&set), url.clone());
1714                let result = set.fetch_and_store_url(url).await;
1715                settle_discovered_url(pending_guard, result);
1716                refresh_sleep = schedule_next_refresh(&set).await;
1717            }
1718        }
1719    }
1720}
1721
1722// Abort-safety guard for the discovery arm in `run_crl_refresher`. Cooperative
1723// shutdown already leaves `pending_urls` consistent because the arm body runs to
1724// a normal `Ok`/`Err` settlement point, but `JoinHandle::abort` drops the future
1725// at whatever `.await` it is currently suspended on. Holding this owned guard
1726// across `fetch_and_store_url` makes that hard-abort path deterministic too: if
1727// the fetch future is dropped before a CRL is confirmed cached, `Drop` removes
1728// the transient in-flight marker so the CDP can be retried by a later handshake.
1729struct PendingUrlGuard {
1730    set: Arc<CrlSet>,
1731    url: String,
1732    armed: bool,
1733}
1734
1735impl PendingUrlGuard {
1736    fn armed(set: Arc<CrlSet>, url: String) -> Self {
1737        Self {
1738            set,
1739            url,
1740            armed: true,
1741        }
1742    }
1743
1744    fn disarm(&mut self) {
1745        self.armed = false;
1746    }
1747}
1748
1749impl Drop for PendingUrlGuard {
1750    fn drop(&mut self) {
1751        if self.armed {
1752            self.set.clear_pending(&self.url);
1753        }
1754    }
1755}
1756
1757fn settle_discovered_url(
1758    mut pending_guard: PendingUrlGuard,
1759    result: Result<bool, RmcpServerKitError>,
1760) {
1761    match result {
1762        // Cached: safe to suppress this URL permanently.
1763        Ok(true) => {
1764            pending_guard.disarm();
1765            pending_guard
1766                .set
1767                .promote_pending_to_seen(&pending_guard.url);
1768        }
1769        // Fetched but refused by the cache cap. Keep the guard armed so scope
1770        // exit clears the in-flight marker before refresh rescheduling, and a
1771        // later handshake can retry; suppressing it here would disable
1772        // revocation for this CDP even though no CRL was ever cached.
1773        Ok(false) => {
1774            tracing::warn!(
1775                url = %pending_guard.url,
1776                "CRL fetched but not admitted to cache (cap reached); will retry on a later handshake"
1777            );
1778        }
1779        Err(error) => {
1780            tracing::warn!(
1781                url = %pending_guard.url,
1782                error = %error,
1783                "CRL discovery fetch failed; will retry on a later handshake"
1784            );
1785        }
1786    }
1787}
1788
1789/// Rebuild the inner rustls verifier from the current CRL cache.
1790///
1791/// # Errors
1792///
1793/// Returns an error if rustls rejects the verifier configuration.
1794pub fn rebuild_verifier<S: std::hash::BuildHasher>(
1795    roots: &Arc<RootCertStore>,
1796    config: &MtlsConfig,
1797    cache: &HashMap<String, CachedCrl, S>,
1798) -> Result<Arc<dyn ClientCertVerifier>, RmcpServerKitError> {
1799    let mut builder = WebPkiClientVerifier::builder(Arc::clone(roots));
1800
1801    if !cache.is_empty() {
1802        let crls = cache
1803            .values()
1804            .map(|cached| cached.der.clone())
1805            .collect::<Vec<_>>();
1806        builder = builder.with_crls(crls);
1807    }
1808    if config.crl_end_entity_only {
1809        builder = builder.only_check_end_entity_revocation();
1810    }
1811    if !config.crl_deny_on_unavailable {
1812        builder = builder.allow_unknown_revocation_status();
1813    }
1814    if config.crl_enforce_expiration {
1815        builder = builder.enforce_revocation_expiration();
1816    }
1817    if !config.required {
1818        builder = builder.allow_unauthenticated();
1819    }
1820
1821    builder
1822        .build()
1823        .map_err(|error| RmcpServerKitError::Tls(format!("mTLS verifier error: {error}")))
1824}
1825
1826/// Parse `thisUpdate` and `nextUpdate` metadata from a DER-encoded CRL.
1827///
1828/// # Errors
1829///
1830/// Returns an error if the CRL cannot be parsed.
1831pub fn parse_crl_metadata(
1832    der: &[u8],
1833) -> Result<(SystemTime, Option<SystemTime>), RmcpServerKitError> {
1834    let (_, crl) = CertificateRevocationList::from_der(der)
1835        .map_err(|error| RmcpServerKitError::Tls(format!("invalid CRL DER: {error:?}")))?;
1836
1837    Ok((
1838        asn1_time_to_system_time(crl.last_update()),
1839        crl.next_update().map(asn1_time_to_system_time),
1840    ))
1841}
1842
1843async fn schedule_next_refresh(set: &CrlSet) -> Pin<Box<Sleep>> {
1844    let duration = next_refresh_delay(set).await;
1845    boxed_sleep(duration)
1846}
1847
1848fn boxed_sleep(duration: Duration) -> Pin<Box<Sleep>> {
1849    Box::pin(tokio::time::sleep_until(Instant::now() + duration))
1850}
1851
1852async fn next_refresh_delay(set: &CrlSet) -> Duration {
1853    if let Some(interval) = set.config.crl_refresh_interval {
1854        return clamp_refresh(interval);
1855    }
1856
1857    let now = SystemTime::now();
1858    let cache = set.cache_lock().read().await;
1859    let mut next = MAX_AUTO_REFRESH;
1860
1861    for cached in cache.values() {
1862        if let Some(next_update) = cached.next_update {
1863            let duration = next_update.duration_since(now).unwrap_or(Duration::ZERO);
1864            next = next.min(clamp_refresh(duration));
1865        }
1866    }
1867    drop(cache);
1868
1869    next
1870}
1871
1872/// Get-or-insert the per-host fetch semaphore for `host_key`.
1873///
1874/// When the map is at `max_host_semaphores`, idle entries (no in-flight
1875/// fetch) are evicted before rejecting, so the cap only fails when `max`
1876/// distinct hosts are *concurrently* fetching — it is never a permanent
1877/// lockout. Every clone of a host semaphore is created while holding the
1878/// map lock, and a clone outlives the critical section only while a fetch
1879/// is in flight, so an entry with `Arc::strong_count == 1` is provably
1880/// idle and safe to drop.
1881fn acquire_host_semaphore(
1882    map: &mut HashMap<String, Arc<Semaphore>>,
1883    host_key: &str,
1884    max_host_semaphores: usize,
1885) -> Result<Arc<Semaphore>, RmcpServerKitError> {
1886    if !map.contains_key(host_key) {
1887        if map.len() >= max_host_semaphores {
1888            // Self-heal: drop semaphores with no in-flight fetch.
1889            map.retain(|_, semaphore| Arc::strong_count(semaphore) > 1);
1890        }
1891        if map.len() >= max_host_semaphores {
1892            return Err(RmcpServerKitError::Config(
1893                "crl_host_semaphore_cap_exceeded: too many distinct CRL hosts in flight".to_owned(),
1894            ));
1895        }
1896        map.insert(host_key.to_owned(), Arc::new(Semaphore::new(1)));
1897    }
1898    match map.get(host_key) {
1899        Some(semaphore) => Ok(Arc::clone(semaphore)),
1900        None => Err(RmcpServerKitError::Tls(
1901            "CRL host semaphore missing after insertion".to_owned(),
1902        )),
1903    }
1904}
1905
1906/// Fetch a single CRL URL through the global + per-host concurrency caps.
1907///
1908/// `global_sem` caps total simultaneous CRL fetches process-wide.
1909/// `host_semaphores` ensures at most one in-flight fetch per origin host
1910/// (an SSRF amplification defense); at the host cap, idle entries are
1911/// evicted on demand. Both permits are dropped when the returned future
1912/// completes (whether `Ok` or `Err`).
1913// cancel-safe for permits: cancelling queued `acquire_owned` loses only queue
1914// position, acquired global/host permits RAII-drop, and host-map insertion can
1915// leave only an idle bounded semaphore entry that later self-heals.
1916async fn gated_fetch(
1917    client: &reqwest::Client,
1918    global_sem: &Arc<Semaphore>,
1919    host_semaphores: &Arc<tokio::sync::Mutex<HashMap<String, Arc<Semaphore>>>>,
1920    url: &str,
1921    allow_http: bool,
1922    max_bytes: u64,
1923    max_host_semaphores: usize,
1924) -> Result<CachedCrl, RmcpServerKitError> {
1925    let host_key = Url::parse(url)
1926        .ok()
1927        .and_then(|u| u.host_str().map(str::to_owned))
1928        .unwrap_or_else(|| url.to_owned());
1929
1930    let host_sem = {
1931        let mut map = host_semaphores.lock().await;
1932        acquire_host_semaphore(&mut map, &host_key, max_host_semaphores)?
1933    };
1934
1935    let _global_permit = Arc::clone(global_sem)
1936        .acquire_owned()
1937        .await
1938        .map_err(|error| {
1939            RmcpServerKitError::Tls(format!("CRL global semaphore closed: {error}"))
1940        })?;
1941    let _host_permit = host_sem
1942        .acquire_owned()
1943        .await
1944        .map_err(|error| RmcpServerKitError::Tls(format!("CRL host semaphore closed: {error}")))?;
1945
1946    fetch_crl(client, url, allow_http, max_bytes).await
1947}
1948
1949// cancel-safe: DNS lookup, request send, chunk reads, DER parse, and metadata
1950// extraction build only a local `CachedCrl`; CRL cache/verifier state changes
1951// happen later, when callers commit the returned value.
1952async fn fetch_crl(
1953    client: &reqwest::Client,
1954    url: &str,
1955    allow_http: bool,
1956    max_bytes: u64,
1957) -> Result<CachedCrl, RmcpServerKitError> {
1958    let parsed = Url::parse(url)
1959        .map_err(|error| RmcpServerKitError::Tls(format!("CRL URL parse {url}: {error}")))?;
1960
1961    if let Err(reason) = check_scheme(&parsed, allow_http) {
1962        // Sanitized: the gate must not echo what it rejects (the URL may
1963        // carry userinfo credentials — the very thing being refused).
1964        let sanitized = sanitized_url_for_log(&parsed);
1965        tracing::warn!(url = %sanitized, reason, "CRL fetch denied: scheme");
1966        return Err(RmcpServerKitError::Tls(format!(
1967            "CRL scheme rejected ({reason}): {sanitized}"
1968        )));
1969    }
1970
1971    let host = parsed
1972        .host_str()
1973        .ok_or_else(|| RmcpServerKitError::Tls(format!("CRL URL has no host: {url}")))?;
1974    let port = parsed
1975        .port_or_known_default()
1976        .ok_or_else(|| RmcpServerKitError::Tls(format!("CRL URL has no known port: {url}")))?;
1977
1978    let addrs = lookup_host((host, port))
1979        .await
1980        .map_err(|error| RmcpServerKitError::Tls(format!("CRL DNS resolution {url}: {error}")))?;
1981
1982    let mut any_addr = false;
1983    for addr in addrs {
1984        any_addr = true;
1985        if let Some(reason) = ip_block_reason(addr.ip()) {
1986            tracing::warn!(
1987                url = %url,
1988                resolved_ip = %addr.ip(),
1989                reason,
1990                "CRL fetch denied: blocked IP"
1991            );
1992            return Err(RmcpServerKitError::Tls(format!(
1993                "CRL host resolved to blocked IP ({reason}): {url}"
1994            )));
1995        }
1996    }
1997    if !any_addr {
1998        return Err(RmcpServerKitError::Tls(format!(
1999            "CRL DNS resolution returned no addresses: {url}"
2000        )));
2001    }
2002
2003    let mut response = client
2004        .get(url)
2005        .send()
2006        .await
2007        .map_err(|error| RmcpServerKitError::Tls(format!("CRL fetch {url}: {error}")))?
2008        .error_for_status()
2009        .map_err(|error| RmcpServerKitError::Tls(format!("CRL fetch {url}: {error}")))?;
2010
2011    // Enforce body cap by streaming chunk-by-chunk; a malicious or
2012    // misconfigured server cannot allocate more than `max_bytes` of memory.
2013    let initial_capacity = usize::try_from(max_bytes.min(64 * 1024)).unwrap_or(64 * 1024);
2014    let mut body: Vec<u8> = Vec::with_capacity(initial_capacity);
2015    while let Some(chunk) = response
2016        .chunk()
2017        .await
2018        .map_err(|error| RmcpServerKitError::Tls(format!("CRL read {url}: {error}")))?
2019    {
2020        let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
2021        let body_len = u64::try_from(body.len()).unwrap_or(u64::MAX);
2022        if body_len.saturating_add(chunk_len) > max_bytes {
2023            return Err(RmcpServerKitError::Tls(format!(
2024                "CRL body exceeded cap of {max_bytes} bytes: {url}"
2025            )));
2026        }
2027        body.extend_from_slice(&chunk);
2028    }
2029
2030    let der = CertificateRevocationListDer::from(body);
2031    let (this_update, next_update) = parse_crl_metadata(der.as_ref())?;
2032
2033    Ok(CachedCrl {
2034        der,
2035        this_update,
2036        next_update,
2037        fetched_at: SystemTime::now(),
2038        source_url: url.to_owned(),
2039    })
2040}
2041
2042fn should_refresh_cached(
2043    cached: &CachedCrl,
2044    now: SystemTime,
2045    fixed_interval: Option<Duration>,
2046) -> bool {
2047    if let Some(interval) = fixed_interval {
2048        return cached
2049            .fetched_at
2050            .checked_add(clamp_refresh(interval))
2051            .is_none_or(|deadline| now >= deadline);
2052    }
2053
2054    cached
2055        .next_update
2056        .is_none_or(|next_update| now >= next_update)
2057}
2058
2059fn clamp_refresh(duration: Duration) -> Duration {
2060    duration.clamp(MIN_AUTO_REFRESH, MAX_AUTO_REFRESH)
2061}
2062
2063/// 9999-12-31T23:59:59Z — the maximum instant expressible as an ASN.1
2064/// GeneralizedTime (four-digit year). Used to clamp absurd positive
2065/// timestamps before converting to [`SystemTime`].
2066const MAX_ASN1_TIMESTAMP_SECS: u64 = 253_402_300_799;
2067
2068/// Convert an ASN.1 time to [`SystemTime`] without ever panicking.
2069///
2070/// CRL metadata is parsed from raw fetched bytes *before* signature
2071/// validation, so timestamps are attacker-controlled. Platform
2072/// `SystemTime` ranges differ (Windows cannot represent pre-1601);
2073/// unrepresentable values are clamped toward [`UNIX_EPOCH`], which is the
2074/// safe direction: it can only make a CRL look *older* (forcing an
2075/// eager refresh), never fresher.
2076fn asn1_time_to_system_time(time: x509_parser::time::ASN1Time) -> SystemTime {
2077    let timestamp = time.timestamp();
2078    if timestamp >= 0 {
2079        let seconds = u64::try_from(timestamp)
2080            .unwrap_or(0)
2081            .min(MAX_ASN1_TIMESTAMP_SECS);
2082        UNIX_EPOCH
2083            .checked_add(Duration::from_secs(seconds))
2084            .unwrap_or(UNIX_EPOCH)
2085    } else {
2086        UNIX_EPOCH
2087            .checked_sub(Duration::from_secs(timestamp.unsigned_abs()))
2088            .unwrap_or(UNIX_EPOCH)
2089    }
2090}
2091
2092#[cfg(test)]
2093mod tests {
2094    #![allow(
2095        deprecated,
2096        reason = "these tests deliberately exercise the deprecated out-of-band cache surface and the ungated test constructors; that is precisely the behaviour under test"
2097    )]
2098
2099    use std::sync::{
2100        Mutex as StdMutex,
2101        atomic::{AtomicBool, AtomicUsize, Ordering},
2102    };
2103
2104    use rcgen::{
2105        BasicConstraints, CertificateParams, CertifiedIssuer, DnType, IsCa, KeyPair,
2106        KeyUsagePurpose,
2107    };
2108
2109    use super::*;
2110
2111    #[derive(Clone, Default)]
2112    struct CapturedLogs(Arc<StdMutex<Vec<u8>>>);
2113
2114    impl CapturedLogs {
2115        fn contents(&self) -> String {
2116            let guard = self
2117                .0
2118                .lock()
2119                .unwrap_or_else(std::sync::PoisonError::into_inner);
2120            String::from_utf8_lossy(&guard).into_owned()
2121        }
2122    }
2123
2124    struct CapturedLogsWriter(Arc<StdMutex<Vec<u8>>>);
2125
2126    impl std::io::Write for CapturedLogsWriter {
2127        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
2128            {
2129                let mut guard = self
2130                    .0
2131                    .lock()
2132                    .unwrap_or_else(std::sync::PoisonError::into_inner);
2133                guard.extend_from_slice(buf);
2134            }
2135            Ok(buf.len())
2136        }
2137
2138        fn flush(&mut self) -> std::io::Result<()> {
2139            Ok(())
2140        }
2141    }
2142
2143    impl<'writer> tracing_subscriber::fmt::MakeWriter<'writer> for CapturedLogs {
2144        type Writer = CapturedLogsWriter;
2145
2146        fn make_writer(&'writer self) -> Self::Writer {
2147            CapturedLogsWriter(Arc::clone(&self.0))
2148        }
2149    }
2150
2151    fn asn1(timestamp: i64) -> x509_parser::time::ASN1Time {
2152        x509_parser::time::ASN1Time::from_timestamp(timestamp).expect("valid ASN.1 timestamp")
2153    }
2154
2155    fn install_ring_provider() {
2156        // `CrlSet::new` builds a reqwest client whose rustls backend is compiled
2157        // with `rustls-no-provider`; installing the provider is idempotent and
2158        // keeps these unit tests independent from whichever integration test
2159        // happens to initialize crypto first.
2160        let _ = rustls::crypto::ring::default_provider().install_default();
2161    }
2162
2163    fn test_ca_root() -> CertificateDer<'static> {
2164        let mut params = CertificateParams::new(Vec::<String>::new()).expect("ca params");
2165        params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
2166        params.key_usages = vec![
2167            KeyUsagePurpose::KeyCertSign,
2168            KeyUsagePurpose::CrlSign,
2169            KeyUsagePurpose::DigitalSignature,
2170        ];
2171        params
2172            .distinguished_name
2173            .push(DnType::CommonName, "mtls-revocation-unit-test-ca");
2174        let key = KeyPair::generate().expect("ca key");
2175        let issuer: CertifiedIssuer<'static, KeyPair> =
2176            CertifiedIssuer::self_signed(params, key).expect("ca self-signed");
2177        issuer.der().clone()
2178    }
2179
2180    fn test_mtls_config() -> MtlsConfig {
2181        serde_json::from_value(serde_json::json!({
2182            "ca_cert_path": "memory://ca.pem",
2183            "required": true,
2184            "default_role": "viewer",
2185            "crl_enabled": true,
2186            "crl_deny_on_unavailable": false,
2187            "crl_allow_http": true,
2188            "crl_enforce_expiration": true,
2189            "crl_end_entity_only": false,
2190            "crl_fetch_timeout": "30s",
2191            "crl_stale_grace": "24h",
2192            "crl_max_concurrent_fetches": 1,
2193            "crl_max_response_bytes": 5_242_880,
2194            "crl_discovery_rate_per_min": 60,
2195            "crl_max_host_semaphores": 16,
2196            "crl_max_seen_urls": 16,
2197            "crl_max_cache_entries": 16,
2198        }))
2199        .expect("verifier mtls config")
2200    }
2201
2202    fn test_crl_set_with_receiver() -> (Arc<CrlSet>, mpsc::UnboundedReceiver<String>) {
2203        test_crl_set_with_receiver_config(test_mtls_config())
2204    }
2205
2206    fn test_crl_set_with_receiver_config(
2207        config: MtlsConfig,
2208    ) -> (Arc<CrlSet>, mpsc::UnboundedReceiver<String>) {
2209        install_ring_provider();
2210        let mut roots = RootCertStore::empty();
2211        roots.add(test_ca_root()).expect("add ca root");
2212        CrlSet::__test_with_kept_receiver(Arc::new(roots), config, vec![])
2213            .expect("empty CRL set with kept receiver")
2214    }
2215
2216    fn pending_contains(set: &CrlSet, url: &str) -> bool {
2217        set.pending_urls
2218            .lock()
2219            .unwrap_or_else(std::sync::PoisonError::into_inner)
2220            .contains(url)
2221    }
2222
2223    fn seen_contains(set: &CrlSet, url: &str) -> bool {
2224        set.seen_urls
2225            .lock()
2226            .unwrap_or_else(std::sync::PoisonError::into_inner)
2227            .contains(url)
2228    }
2229
2230    fn mark_pending(set: &CrlSet, url: &str) {
2231        let mut pending = set
2232            .pending_urls
2233            .lock()
2234            .unwrap_or_else(std::sync::PoisonError::into_inner);
2235        pending.insert(url.to_owned());
2236    }
2237
2238    /// The pending marker must exist before a URL becomes observable on the
2239    /// discovery channel; otherwise a settlement racing the send strands the
2240    /// URL as permanently pending.
2241    ///
2242    /// Asserted at the exact instant it can be violated, via the pre-send
2243    /// probe. The previous form spun a contending thread across 200 attempts
2244    /// and could pass without ever producing the bad interleaving.
2245    #[test]
2246    fn discovery_does_not_send_before_pending_marker_exists() {
2247        let mut config = test_mtls_config();
2248        config.crl_discovery_rate_per_min = 10_000;
2249        config.crl_max_seen_urls = 512;
2250        let (set, mut discover_rx) = test_crl_set_with_receiver_config(config);
2251
2252        let url = "http://pending-order.example.test/crl".to_owned();
2253        let observed = Arc::new(AtomicUsize::new(0));
2254        let probe_observed = Arc::clone(&observed);
2255        let probe_url = url.clone();
2256
2257        set.__test_set_discovery_send_probe(Arc::new(move |set: &CrlSet, sent: &str| {
2258            assert_eq!(sent, probe_url, "probe must observe the discovered URL");
2259            assert!(
2260                pending_contains(set, sent),
2261                "URL {sent} became observable before its pending marker existed"
2262            );
2263            probe_observed.fetch_add(1, Ordering::Relaxed);
2264        }));
2265
2266        let _ = set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]);
2267
2268        assert_eq!(
2269            discover_rx.try_recv().ok().as_deref(),
2270            Some(url.as_str()),
2271            "the discovered URL must be published exactly once"
2272        );
2273        assert_eq!(
2274            observed.load(Ordering::Relaxed),
2275            1,
2276            "the pre-send probe must have fired for this URL"
2277        );
2278
2279        set.__test_settle_pending(&url, false);
2280        assert!(
2281            !pending_contains(&set, &url),
2282            "settling a fetch must not strand pending URL {url}"
2283        );
2284        assert!(
2285            !set.__test_is_seen(&url),
2286            "settling a failed fetch must leave URL {url} discoverable again"
2287        );
2288    }
2289
2290    /// Covers the cap-before-publication invariant directly rather than by
2291    /// driving `bootstrap_fetch`, because `bootstrap_fetch` cannot be reached
2292    /// from a test without weakening production SSRF screening.
2293    ///
2294    /// `fetch_crl` performs its own DNS precheck -- `tokio::net::lookup_host`
2295    /// followed by `ip_block_reason` -- *before* handing the request to
2296    /// `reqwest`. That precheck is independent of the `dns_resolver` injected
2297    /// into the client, so a custom resolver cannot redirect a fetch to a
2298    /// local mock: a `wiremock` server on loopback is rejected as a blocked IP
2299    /// before `reqwest` is ever invoked.
2300    ///
2301    /// Reaching `bootstrap_fetch` end to end would therefore require either a
2302    /// test-only bypass of that precheck, or a CRL served from a public
2303    /// non-private address. The first is deliberately not done -- a bypass
2304    /// seam next to CRL fetching is a worse security liability than the
2305    /// coverage it would buy -- and the second is not available offline.
2306    ///
2307    /// This test asserts the property that actually matters: the cap is
2308    /// applied *before* `CrlSet::new`, so `cache`, `cached_urls`, and
2309    /// `inner_verifier` are all derived from one already-bounded map. Post-hoc
2310    /// mutation would violate the publication ordering documented above.
2311    ///
2312    /// If you are here to "improve" this by making it drive `bootstrap_fetch`:
2313    /// do not add a precheck bypass to `fetch_crl` to do it.
2314    #[tokio::test]
2315    async fn bootstrap_cache_cap_is_applied_before_crl_set_publication() {
2316        let cap = 4usize;
2317        let mut config = test_mtls_config();
2318        config.crl_max_cache_entries = cap;
2319        let (discover_tx, _discover_rx) = mpsc::unbounded_channel();
2320        install_ring_provider();
2321        let mut roots = RootCertStore::empty();
2322        roots.add(test_ca_root()).expect("add ca root");
2323        let roots = Arc::new(roots);
2324        let now = SystemTime::now();
2325        let initial_cache: HashMap<String, CachedCrl> = (0..cap + 3)
2326            .rev()
2327            .map(|index| format!("https://bootstrap-{index:02}.example.test/crl"))
2328            .map(|url| {
2329                let mut cached = CachedCrl::__test_synthetic(now);
2330                cached.source_url = url.clone();
2331                (url, cached)
2332            })
2333            .collect();
2334
2335        let set = new_crl_set_from_bootstrap_cache(roots, config, discover_tx, initial_cache)
2336            .expect("bootstrap cache should build CRL set");
2337        let cache_keys = {
2338            let cache = set.cache_lock().read().await;
2339            assert_eq!(cache.len(), cap, "bootstrap cache len must be capped");
2340            cache.keys().cloned().collect::<HashSet<_>>()
2341        };
2342        let cached_url_keys = {
2343            let cached_urls = &set.verifier_state.load().cached_urls;
2344            assert_eq!(
2345                cached_urls.len(),
2346                cap,
2347                "cached_urls len must match capped bootstrap cache"
2348            );
2349            cached_urls.iter().cloned().collect::<HashSet<_>>()
2350        };
2351        let expected: HashSet<String> = (0..cap)
2352            .map(|index| format!("https://bootstrap-{index:02}.example.test/crl"))
2353            .collect();
2354
2355        assert_eq!(
2356            cache_keys, cached_url_keys,
2357            "bootstrap cache and cached_urls must publish the same key set"
2358        );
2359        assert_eq!(
2360            cache_keys, expected,
2361            "bootstrap admission must keep the first cap URLs in sort order"
2362        );
2363    }
2364
2365    async fn wait_for_host_fetch_to_block_on_global_permit(set: &CrlSet, host: &str) {
2366        tokio::time::timeout(Duration::from_secs(2), async {
2367            loop {
2368                if set.host_semaphores.lock().await.contains_key(host) {
2369                    return;
2370                }
2371                tokio::task::yield_now().await;
2372            }
2373        })
2374        .await
2375        .expect("refresher must reach the CRL fetch path before abort");
2376    }
2377
2378    #[tokio::test]
2379    async fn aborted_refresher_does_not_strand_pending_url() {
2380        let (set, discover_rx) = test_crl_set_with_receiver();
2381        let url = "http://abort.example.test/crl";
2382        let host = "abort.example.test";
2383        let held_global_permit = Arc::clone(&set.global_fetch_sem)
2384            .acquire_owned()
2385            .await
2386            .expect("test semaphore is open");
2387
2388        assert!(!pending_contains(&set, url));
2389        assert!(!seen_contains(&set, url));
2390
2391        let _ = set.__test_note_discovered_urls_by_cert(&[url.to_owned()], &[]);
2392        assert!(
2393            pending_contains(&set, url),
2394            "queued URL must be marked in-flight before the fetch starts"
2395        );
2396        assert!(
2397            !seen_contains(&set, url),
2398            "queueing alone must not promote to the permanent dedup set"
2399        );
2400
2401        let handle = tokio::spawn(run_crl_refresher(
2402            Arc::clone(&set),
2403            discover_rx,
2404            CancellationToken::new(),
2405        ));
2406
2407        wait_for_host_fetch_to_block_on_global_permit(&set, host).await;
2408        handle.abort();
2409        let join_error = handle
2410            .await
2411            .expect_err("aborted refresher must not complete normally");
2412        assert!(join_error.is_cancelled());
2413        drop(held_global_permit);
2414
2415        assert!(
2416            !pending_contains(&set, url),
2417            "aborting while fetch_and_store_url awaits must clear the in-flight marker"
2418        );
2419        assert!(
2420            !seen_contains(&set, url),
2421            "an aborted fetch must not promote the URL to the permanent dedup set"
2422        );
2423
2424        let _ = set.__test_note_discovered_urls(&[url.to_owned()]);
2425        assert!(
2426            pending_contains(&set, url),
2427            "once the stale marker is gone, the same URL can be queued again"
2428        );
2429        assert!(
2430            !seen_contains(&set, url),
2431            "retry admission must still be pending-only, not permanent suppression"
2432        );
2433    }
2434
2435    #[test]
2436    fn discovered_url_settlement_promotes_only_confirmed_cache_admission() {
2437        let (set, _discover_rx) = test_crl_set_with_receiver();
2438        let url = "http://settle-ok.example.test/crl";
2439        mark_pending(&set, url);
2440
2441        let pending_guard = PendingUrlGuard::armed(Arc::clone(&set), url.to_owned());
2442        settle_discovered_url(pending_guard, Ok(true));
2443
2444        assert!(
2445            seen_contains(&set, url),
2446            "Ok(true) means the CRL is cached and must permanently dedup the URL"
2447        );
2448        assert!(
2449            !pending_contains(&set, url),
2450            "promotion must remove the transient in-flight marker"
2451        );
2452    }
2453
2454    #[test]
2455    fn discovered_url_settlement_clears_cache_cap_rejection_and_warns() {
2456        let (set, _discover_rx) = test_crl_set_with_receiver();
2457        let url = "http://settle-cap.example.test/crl";
2458        mark_pending(&set, url);
2459        let logs = CapturedLogs::default();
2460        let subscriber = tracing_subscriber::fmt()
2461            .with_max_level(tracing::Level::WARN)
2462            .with_writer(logs.clone())
2463            .with_ansi(false)
2464            .without_time()
2465            .finish();
2466        let _guard = tracing::subscriber::set_default(subscriber);
2467
2468        let pending_guard = PendingUrlGuard::armed(Arc::clone(&set), url.to_owned());
2469        settle_discovered_url(pending_guard, Ok(false));
2470
2471        assert!(
2472            !pending_contains(&set, url),
2473            "cache-cap rejection must leave the URL retriable"
2474        );
2475        assert!(
2476            !seen_contains(&set, url),
2477            "cache-cap rejection must not promote permanent suppression"
2478        );
2479        let contents = logs.contents();
2480        assert!(
2481            contents.contains(
2482                "CRL fetched but not admitted to cache (cap reached); will retry on a later handshake"
2483            ),
2484            "existing cache-cap warning must still be emitted: {contents}"
2485        );
2486    }
2487
2488    #[test]
2489    fn discovered_url_settlement_clears_fetch_failure_and_warns() {
2490        let (set, _discover_rx) = test_crl_set_with_receiver();
2491        let url = "http://settle-error.example.test/crl";
2492        mark_pending(&set, url);
2493        let logs = CapturedLogs::default();
2494        let subscriber = tracing_subscriber::fmt()
2495            .with_max_level(tracing::Level::WARN)
2496            .with_writer(logs.clone())
2497            .with_ansi(false)
2498            .without_time()
2499            .finish();
2500        let _guard = tracing::subscriber::set_default(subscriber);
2501
2502        let pending_guard = PendingUrlGuard::armed(Arc::clone(&set), url.to_owned());
2503        settle_discovered_url(
2504            pending_guard,
2505            Err(RmcpServerKitError::Tls("test-fetch-failed".to_owned())),
2506        );
2507
2508        assert!(
2509            !pending_contains(&set, url),
2510            "fetch failure must leave the URL retriable"
2511        );
2512        assert!(
2513            !seen_contains(&set, url),
2514            "fetch failure must not promote permanent suppression"
2515        );
2516        let contents = logs.contents();
2517        assert!(
2518            contents.contains("CRL discovery fetch failed; will retry on a later handshake"),
2519            "existing fetch-failure warning must still be emitted: {contents}"
2520        );
2521        assert!(
2522            contents.contains("test-fetch-failed"),
2523            "existing warning must still include the fetch error: {contents}"
2524        );
2525    }
2526
2527    /// The userinfo gate fires before DNS resolution (no network needed)
2528    /// and the surfaced error must not echo the rejected credentials.
2529    #[tokio::test]
2530    async fn fetch_crl_rejects_userinfo_without_echoing_credentials() {
2531        // reqwest with `rustls-no-provider` requires a process-wide crypto
2532        // provider before any Client is built (same pattern as the
2533        // transport/oauth test suites).
2534        let _ = rustls::crypto::ring::default_provider().install_default();
2535        let client = reqwest::Client::new();
2536        let err = fetch_crl(&client, "https://u:p@crl.example/ca.crl", false, 1024)
2537            .await
2538            .expect_err("userinfo-bearing CRL URL must be rejected");
2539        let rendered = err.to_string();
2540        assert!(
2541            rendered.contains("userinfo_forbidden"),
2542            "error must carry the rejection reason: {rendered}"
2543        );
2544        assert!(
2545            !rendered.contains("u:p"),
2546            "error must not echo the rejected credentials: {rendered}"
2547        );
2548    }
2549
2550    /// `extract_cdp_urls`'s scheme/userinfo guard reuses the same gate;
2551    /// the sanitizer keeps credentials out of its debug logging too.
2552    #[test]
2553    fn sanitizer_used_by_rejection_sites_strips_credentials() {
2554        let parsed = Url::parse("https://u:p@crl.example/ca.crl").expect("parse");
2555        let sanitized = sanitized_url_for_log(&parsed);
2556        assert_eq!(sanitized, "https://crl.example");
2557        assert!(!sanitized.contains("u:p"));
2558    }
2559
2560    #[test]
2561    fn asn1_time_clamps_unrepresentable_timestamps() {
2562        // Year 1500 — pre-1601, NOT representable by Windows `SystemTime`.
2563        // Pre-fix this panicked on Windows; now it must return a value no
2564        // later than the epoch on every platform (clamped to UNIX_EPOCH on
2565        // Windows, the real instant on platforms that can represent it).
2566        let year_1500 = asn1_time_to_system_time(asn1(-14_831_769_600));
2567        assert!(year_1500 <= UNIX_EPOCH);
2568        #[cfg(windows)]
2569        assert_eq!(year_1500, UNIX_EPOCH);
2570
2571        // 1601-01-01T00:00:00Z — the exact Windows epoch boundary, which IS
2572        // representable everywhere. No clamp, no panic.
2573        let year_1601 = asn1_time_to_system_time(asn1(-11_644_473_600));
2574        assert!(year_1601 <= UNIX_EPOCH);
2575
2576        // Mildly negative (pre-1970) stays at-or-before the epoch.
2577        assert!(asn1_time_to_system_time(asn1(-2)) <= UNIX_EPOCH);
2578
2579        // Normal positive timestamps round-trip exactly.
2580        assert_eq!(
2581            asn1_time_to_system_time(asn1(1_700_000_000)),
2582            UNIX_EPOCH + Duration::from_secs(1_700_000_000)
2583        );
2584
2585        // The ASN.1 maximum (9999-12-31) is representable and preserved.
2586        let max = i64::try_from(MAX_ASN1_TIMESTAMP_SECS).expect("fits in i64");
2587        assert_eq!(
2588            asn1_time_to_system_time(asn1(max)),
2589            UNIX_EPOCH + Duration::from_secs(MAX_ASN1_TIMESTAMP_SECS)
2590        );
2591    }
2592
2593    #[test]
2594    fn host_semaphore_evicts_idle_at_cap() {
2595        let mut map = HashMap::new();
2596        for i in 0..4 {
2597            // Dropped immediately: only the map holds each semaphore (idle).
2598            drop(
2599                acquire_host_semaphore(&mut map, &format!("idle-{i}.example"), 4)
2600                    .expect("under cap"),
2601            );
2602        }
2603        assert_eq!(map.len(), 4);
2604
2605        // At the cap, a NEW host must succeed by evicting idle entries —
2606        // the cap error is not sticky.
2607        let sem = acquire_host_semaphore(&mut map, "new-host.example", 4)
2608            .expect("idle eviction frees space for a new host");
2609        assert!(map.contains_key("new-host.example"));
2610        drop(sem);
2611    }
2612
2613    #[test]
2614    fn host_semaphore_keeps_inflight_at_cap() {
2615        let mut map = HashMap::new();
2616        // Held across the cap check: simulates an in-flight fetch.
2617        let inflight = acquire_host_semaphore(&mut map, "busy.example", 3).expect("under cap");
2618        for i in 0..2 {
2619            drop(
2620                acquire_host_semaphore(&mut map, &format!("idle-{i}.example"), 3)
2621                    .expect("under cap"),
2622            );
2623        }
2624        assert_eq!(map.len(), 3);
2625
2626        drop(
2627            acquire_host_semaphore(&mut map, "new-host.example", 3)
2628                .expect("idle entries evicted while in-flight survives"),
2629        );
2630        assert!(
2631            map.contains_key("busy.example"),
2632            "in-flight host must survive eviction"
2633        );
2634        assert!(map.contains_key("new-host.example"));
2635        drop(inflight);
2636    }
2637
2638    #[test]
2639    fn host_semaphore_cap_error_when_all_inflight() {
2640        let mut map = HashMap::new();
2641        let held: Vec<_> = (0..2)
2642            .map(|i| {
2643                acquire_host_semaphore(&mut map, &format!("busy-{i}.example"), 2)
2644                    .expect("under cap")
2645            })
2646            .collect();
2647
2648        let result = acquire_host_semaphore(&mut map, "new-host.example", 2);
2649        assert!(
2650            result.is_err(),
2651            "cap must still reject when every entry has an in-flight fetch"
2652        );
2653        drop(held);
2654    }
2655
2656    // ---- CrlSet cache invariant (A1/A2/A3) --------------------------------
2657
2658    fn tamper_test_config() -> MtlsConfig {
2659        let mut config = test_mtls_config();
2660        config.crl_deny_on_unavailable = true;
2661        config.crl_end_entity_only = false;
2662        config.crl_discovery_rate_per_min = 10_000;
2663        config.crl_max_seen_urls = 4096;
2664        config.crl_max_cache_entries = 4096;
2665        config
2666    }
2667
2668    fn synthetic_entry(now: SystemTime) -> CachedCrl {
2669        CachedCrl::__test_synthetic(now)
2670    }
2671
2672    fn identity_of(set: &CrlSet, url: &str) -> Option<EntryIdentity> {
2673        set.verifier_state
2674            .load()
2675            .committed_identities
2676            .get(url)
2677            .cloned()
2678    }
2679
2680    fn warned(set: &CrlSet, which: &str) -> bool {
2681        set.last_cap_warn
2682            .lock()
2683            .unwrap_or_else(std::sync::PoisonError::into_inner)
2684            .contains_key(which)
2685    }
2686
2687    /// A replacement that is indistinguishable from `entry` on every cheap
2688    /// field: same DER length, same 32-byte head and tail, same scalars, same
2689    /// `source_url`. Only a middle byte differs, so only the reallocation can
2690    /// betray it. An implementation comparing anything less than the full
2691    /// identity tuple fails the tests that use this.
2692    fn same_shape_replacement(entry: &CachedCrl) -> CachedCrl {
2693        let mut bytes = entry.der.as_ref().to_vec();
2694        let middle = bytes.len() / 2;
2695        if let Some(byte) = bytes.get_mut(middle) {
2696            *byte ^= 0xFF;
2697        }
2698        CachedCrl {
2699            der: CertificateRevocationListDer::from(bytes),
2700            this_update: entry.this_update,
2701            next_update: entry.next_update,
2702            fetched_at: entry.fetched_at,
2703            source_url: entry.source_url.clone(),
2704        }
2705    }
2706
2707    fn crl_set_with_cached_urls(config: MtlsConfig, count: usize) -> (Arc<CrlSet>, Vec<String>) {
2708        install_ring_provider();
2709        let mut roots = RootCertStore::empty();
2710        roots.add(test_ca_root()).expect("add ca root");
2711        let now = SystemTime::now();
2712        let mut initial_cache = HashMap::new();
2713        let mut urls = Vec::with_capacity(count);
2714        for index in 0..count {
2715            let url = format!("https://cdp-{index:03}.example.test/crl");
2716            initial_cache.insert(url.clone(), synthetic_entry(now));
2717            urls.push(url);
2718        }
2719        let (discover_tx, discover_rx) = mpsc::unbounded_channel();
2720        drop(discover_rx);
2721        let set = CrlSet::new(Arc::new(roots), config, discover_tx, initial_cache)
2722            .expect("crl set with prepopulated cache");
2723        urls.sort();
2724        (set, urls)
2725    }
2726
2727    #[tokio::test]
2728    async fn identity_index_is_seeded_by_new_and_maintained_by_commit() {
2729        let boot = "https://cdp-000.example.test/crl";
2730        let (set, _urls) = crl_set_with_cached_urls(tamper_test_config(), 1);
2731
2732        // Blocker 1 regression: identities seeded only on commit would read
2733        // every bootstrap-fetched CRL as mutated and fail mTLS closed at
2734        // startup.
2735        assert!(
2736            identity_of(&set, boot).is_some(),
2737            "CrlSet::new must record identities for the bootstrap cache"
2738        );
2739
2740        let added = "https://added.example.test/crl";
2741        let now = SystemTime::now();
2742        set.__test_insert_cache(added, synthetic_entry(now)).await;
2743        let added_identity = identity_of(&set, added).expect("commit must record an identity");
2744
2745        set.__test_insert_cache(added, synthetic_entry(now + Duration::from_secs(3_600)))
2746            .await;
2747        assert!(
2748            identity_of(&set, added).as_ref() != Some(&added_identity),
2749            "legitimate replacement must change the recorded identity"
2750        );
2751
2752        set.commit_cache_update_atomically(Vec::new(), &[added.to_owned()])
2753            .await
2754            .expect("removal commit");
2755        assert!(
2756            identity_of(&set, added).is_none(),
2757            "removal must drop the identity in the same publication"
2758        );
2759    }
2760
2761    #[test]
2762    fn identity_covers_der_bytes_beyond_the_sampled_head_and_tail() {
2763        let entry = synthetic_entry(SystemTime::now());
2764        assert!(
2765            entry.der.as_ref().len() > 64,
2766            "fixture must be longer than head+tail so the middle is unsampled"
2767        );
2768        assert!(
2769            entry_identity(&entry) != entry_identity(&same_shape_replacement(&entry)),
2770            "a same-length middle-byte edit must still change the identity"
2771        );
2772    }
2773
2774    #[test]
2775    fn identity_covers_every_scalar_field() {
2776        let now = SystemTime::now();
2777        let base = synthetic_entry(now);
2778        let baseline = entry_identity(&base);
2779
2780        let mut this_update = base.clone();
2781        this_update.this_update = now + Duration::from_secs(1);
2782        let mut next_update = base.clone();
2783        next_update.next_update = None;
2784        let mut fetched_at = base.clone();
2785        fetched_at.fetched_at = now + Duration::from_secs(1);
2786        let mut source_url = base;
2787        source_url.source_url = "test://other".to_owned();
2788
2789        for (label, mutated) in [
2790            ("this_update", this_update),
2791            ("next_update", next_update),
2792            ("fetched_at", fetched_at),
2793            ("source_url", source_url),
2794        ] {
2795            assert!(
2796                entry_identity(&mutated) != baseline,
2797                "mutating {label} alone must change the identity"
2798            );
2799        }
2800    }
2801
2802    #[tokio::test]
2803    async fn precheck_denies_after_same_key_replace_through_public_cache() {
2804        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2805        let url = "https://replace.example.test/crl".to_owned();
2806        let now = SystemTime::now();
2807
2808        let committed = synthetic_entry(now);
2809        set.__test_insert_cache(&url, committed.clone()).await;
2810        assert!(
2811            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
2812            "a legitimately committed CRL must admit the handshake"
2813        );
2814
2815        // The replacement matches on DER length, head, tail, every scalar and
2816        // `source_url`. An implementation that compares anything less than the
2817        // full identity tuple admits here and fails this test.
2818        set.__test_replace_cache_entry_unverified(&url, same_shape_replacement(&committed))
2819            .await;
2820
2821        assert!(
2822            set.verifier_state.load().cached_urls.contains(&url),
2823            "precondition: cached_urls must still claim coverage, or the denial proves nothing"
2824        );
2825        assert!(
2826            set.cache_lock().read().await.contains_key(&url),
2827            "precondition: the entry must still be present, so this is a REPLACE and not a removal"
2828        );
2829        assert!(
2830            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
2831            "REPLACE through the public cache leaves cached_urls claiming coverage the verifier does not enforce; it must deny"
2832        );
2833    }
2834
2835    #[tokio::test]
2836    async fn precheck_denies_after_direct_removal_through_public_cache() {
2837        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2838        let url = "https://remove.example.test/crl".to_owned();
2839
2840        set.__test_insert_cache(&url, synthetic_entry(SystemTime::now()))
2841            .await;
2842        assert!(!set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]));
2843
2844        set.cache_lock().write().await.remove(&url);
2845
2846        assert!(
2847            set.verifier_state.load().cached_urls.contains(&url),
2848            "precondition: cached_urls must still claim the removed URL"
2849        );
2850        assert!(
2851            !set.cache_lock().read().await.contains_key(&url),
2852            "precondition: the live entry must actually be gone"
2853        );
2854        assert!(
2855            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
2856            "cached_urls still claims a URL whose entry was removed out of band; it must deny"
2857        );
2858    }
2859
2860    #[tokio::test]
2861    async fn precheck_uses_committed_state_when_cache_lock_is_temporarily_unavailable() {
2862        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2863        let cached = "https://locked.example.test/crl".to_owned();
2864        let uncached = "https://locked-uncached.example.test/crl".to_owned();
2865
2866        set.__test_insert_cache(&cached, synthetic_entry(SystemTime::now()))
2867            .await;
2868        assert!(!set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&cached), &[]));
2869
2870        // Contention is not tamper. `tokio::sync::RwLock` is write-preferring,
2871        // so denying here would make every legitimate refresh commit deny
2872        // concurrent handshakes; enforcement still runs against the immutable
2873        // committed state, which is what makes falling through sound.
2874        let guard = set.cache_lock().write().await;
2875        assert!(
2876            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&cached), &[]),
2877            "temporary lock contention must not create a spurious denial"
2878        );
2879        assert!(
2880            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&uncached), &[]),
2881            "lock contention must not invent coverage absent from the committed cached_urls"
2882        );
2883        drop(guard);
2884    }
2885
2886    #[tokio::test]
2887    async fn precheck_clean_path_is_unchanged() {
2888        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2889        let cached = "https://cached.example.test/crl".to_owned();
2890        let uncached = "https://uncached.example.test/crl".to_owned();
2891
2892        set.__test_insert_cache(&cached, synthetic_entry(SystemTime::now()))
2893            .await;
2894
2895        assert!(
2896            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&cached), &[]),
2897            "an untampered cached CDP must still admit"
2898        );
2899        assert!(
2900            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&uncached), &[]),
2901            "an uncached CDP must still follow the all(not cached) predicate"
2902        );
2903        assert!(
2904            !set.__test_note_discovered_urls_by_cert(&[cached, uncached], &[]),
2905            "one cached mirror is sufficient coverage (RFC 5280 4.2.1.13)"
2906        );
2907    }
2908
2909    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
2910    async fn concurrent_commits_lose_no_url_and_publish_a_matching_coverage_hint() {
2911        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2912        let now = SystemTime::now();
2913        let total = 64usize;
2914
2915        // Without `commit_lock` these commits each snapshot the same old cache
2916        // and clobber one another, so URLs are silently lost.
2917        let mut tasks = JoinSet::new();
2918        for index in 0..total {
2919            let set = Arc::clone(&set);
2920            tasks.spawn(async move {
2921                set.__test_insert_cache(
2922                    &format!("https://concurrent-{index:03}.example.test/crl"),
2923                    synthetic_entry(now),
2924                )
2925                .await;
2926            });
2927        }
2928        while tasks.join_next().await.is_some() {}
2929
2930        let cache_keys = set
2931            .cache_lock()
2932            .read()
2933            .await
2934            .keys()
2935            .cloned()
2936            .collect::<HashSet<_>>();
2937        assert_eq!(
2938            cache_keys.len(),
2939            total,
2940            "concurrent commits must not lose entries"
2941        );
2942        assert_eq!(
2943            cache_keys,
2944            set.verifier_state.load().cached_urls.clone(),
2945            "the published coverage hint must exactly match the committed cache"
2946        );
2947    }
2948
2949    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
2950    async fn legitimate_refresh_is_never_observed_as_tampering() {
2951        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
2952        let url = "https://coherent.example.test/crl".to_owned();
2953        set.__test_insert_cache(&url, synthetic_entry(SystemTime::now()))
2954            .await;
2955
2956        let stop = Arc::new(AtomicBool::new(false));
2957        let writer_set = Arc::clone(&set);
2958        let writer_url = url.clone();
2959        let writer_stop = Arc::clone(&stop);
2960        let writer = tokio::spawn(async move {
2961            for round in 0..400u64 {
2962                if writer_stop.load(Ordering::Relaxed) {
2963                    break;
2964                }
2965                writer_set
2966                    .__test_insert_cache(
2967                        &writer_url,
2968                        synthetic_entry(SystemTime::now() + Duration::from_secs(round)),
2969                    )
2970                    .await;
2971                tokio::task::yield_now().await;
2972            }
2973            writer_stop.store(true, Ordering::Relaxed);
2974        });
2975
2976        let reader_set = Arc::clone(&set);
2977        let reader_url = url.clone();
2978        let reader_stop = Arc::clone(&stop);
2979        let (denials, checks) = tokio::task::spawn_blocking(move || {
2980            let mut denials = 0usize;
2981            let mut checks = 0usize;
2982            while !reader_stop.load(Ordering::Relaxed) || checks < 1_000 {
2983                if reader_set
2984                    .__test_note_discovered_urls_by_cert(std::slice::from_ref(&reader_url), &[])
2985                {
2986                    denials += 1;
2987                }
2988                checks += 1;
2989                if checks > 200_000 {
2990                    break;
2991                }
2992            }
2993            (denials, checks)
2994        })
2995        .await
2996        .expect("reader task");
2997
2998        writer.await.expect("writer task");
2999        assert_eq!(
3000            denials, 0,
3001            "a legitimate refresh must never be reported as tampering, and must never be observed half-applied"
3002        );
3003        assert!(
3004            checks >= 1_000,
3005            "the reader must actually exercise the precheck: only {checks} checks ran"
3006        );
3007        assert!(
3008            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
3009            "the URL must still admit once churn stops"
3010        );
3011        assert_eq!(
3012            set.cache_lock()
3013                .read()
3014                .await
3015                .keys()
3016                .cloned()
3017                .collect::<HashSet<_>>(),
3018            set.verifier_state.load().cached_urls.clone(),
3019            "cache and published coverage hint must agree after churn"
3020        );
3021    }
3022
3023    /// The commit path must reach `rebuild_verifier` BEFORE taking the cache
3024    /// write lock.
3025    ///
3026    /// Asserted by ordering, not by timing: a held read guard excludes writers,
3027    /// so a commit that surfaces a rebuild error *while that guard is still
3028    /// held* provably performed the rebuild off-lock. Moving the rebuild back
3029    /// inside the write-lock block makes this block until the timeout fires.
3030    /// The previous form of this test compared `try_read` hit ratios, which
3031    /// measured scheduler fairness under CPU oversubscription and failed
3032    /// intermittently on shared CI runners.
3033    ///
3034    /// A multi-threaded runtime is REQUIRED: the commit must be able to make
3035    /// progress on another worker while this task holds the read guard. Under
3036    /// `current_thread` the spawned commit cannot advance and the timeout fires
3037    /// spuriously. Two workers suffice; more only reintroduces oversubscription.
3038    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
3039    async fn commit_does_not_hold_the_cache_write_lock_across_the_verifier_rebuild() {
3040        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
3041        let url = "https://invalid-rebuild.example.test/crl";
3042        let now = SystemTime::now();
3043
3044        let invalid = CachedCrl {
3045            der: CertificateRevocationListDer::from(vec![0_u8]),
3046            this_update: now,
3047            next_update: now.checked_add(Duration::from_secs(24 * 60 * 60)),
3048            fetched_at: now,
3049            source_url: url.to_owned(),
3050        };
3051
3052        let invalid_cache = HashMap::from([(url.to_owned(), invalid.clone())]);
3053        assert!(
3054            rebuild_verifier(&set.roots, &set.config, &invalid_cache).is_err(),
3055            "precondition: the synthetic CRL must make rebuild_verifier fail"
3056        );
3057
3058        let read_guard = set.cache_lock().read().await;
3059
3060        let commit = {
3061            let set = Arc::clone(&set);
3062            tokio::spawn(async move { set.__test_try_insert_cache(url, invalid).await })
3063        };
3064
3065        let finished_while_reader_held = tokio::time::timeout(Duration::from_secs(5), commit).await;
3066
3067        drop(read_guard);
3068
3069        let commit_result = finished_while_reader_held
3070            .expect("commit must reach rebuild_verifier before waiting for the cache write lock")
3071            .expect("commit task must not panic");
3072
3073        assert!(
3074            commit_result.is_err(),
3075            "invalid CRL must fail during verifier rebuild"
3076        );
3077        assert!(
3078            !set.__test_cache_contains(url),
3079            "failed commit must not publish the invalid CRL into the live cache"
3080        );
3081        assert!(
3082            !set.__test_cached_url_contains(url),
3083            "failed commit must not publish invalid CRL coverage into verifier_state"
3084        );
3085    }
3086
3087    // ---- B3: per-handshake CDP URL cap ------------------------------------
3088
3089    fn fail_open_config() -> MtlsConfig {
3090        let mut config = tamper_test_config();
3091        config.crl_deny_on_unavailable = false;
3092        config
3093    }
3094
3095    #[test]
3096    fn bootstrap_urls_are_capped_to_the_cache_limit() {
3097        let cap = 4usize;
3098        let mut urls: Vec<String> = (0..cap + 9)
3099            .map(|index| format!("https://ca-{index:02}.example.test/crl"))
3100            .collect();
3101
3102        cap_bootstrap_urls(&mut urls, cap);
3103
3104        assert_eq!(
3105            urls.len(),
3106            cap,
3107            "a broad CA bundle must not spawn one fetch task per advertised CDP"
3108        );
3109    }
3110
3111    #[test]
3112    fn bootstrap_urls_below_the_cap_are_untouched() {
3113        let mut urls: Vec<String> = (0..3)
3114            .map(|index| format!("https://ca-{index:02}.example.test/crl"))
3115            .collect();
3116        let before = urls.clone();
3117
3118        cap_bootstrap_urls(&mut urls, 16);
3119
3120        assert_eq!(urls, before, "capping must not perturb an in-bounds chain");
3121    }
3122
3123    #[tokio::test]
3124    async fn end_entity_only_mode_does_not_discover_uncapped_intermediate_cdps() {
3125        let mut config = tamper_test_config();
3126        config.crl_end_entity_only = true;
3127        let (set, mut rx) = test_crl_set_with_receiver_config(config);
3128
3129        let end_entity = vec!["https://ee.example.test/crl".to_owned()];
3130        let intermediate: Vec<String> = (0..MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE + 10)
3131            .map(|index| format!("https://int-{index:03}.example.test/crl"))
3132            .collect();
3133
3134        let _ = set.__test_note_discovered_urls_by_cert(&end_entity, &intermediate);
3135
3136        let mut enqueued = Vec::new();
3137        while let Ok(url) = rx.try_recv() {
3138            enqueued.push(url);
3139        }
3140
3141        assert_eq!(
3142            enqueued, end_entity,
3143            "under crl_end_entity_only the capped end-entity set is the only \
3144             discovery source; intermediate CDPs bypass MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE \
3145             and must never be enqueued"
3146        );
3147    }
3148
3149    #[test]
3150    fn cdp_cap_admits_at_the_cap_and_denies_above_it_fail_closed() {
3151        let (at_cap, urls) =
3152            crl_set_with_cached_urls(tamper_test_config(), MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE);
3153        assert!(
3154            !at_cap.__test_note_discovered_urls_by_cert(&urls, &[]),
3155            "exactly the cap must admit; all URLs are cached so nothing else can deny"
3156        );
3157
3158        let (over_cap, urls) = crl_set_with_cached_urls(
3159            tamper_test_config(),
3160            MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE + 1,
3161        );
3162        assert!(
3163            over_cap.__test_note_discovered_urls_by_cert(&urls, &[]),
3164            "one URL past the cap must deny even though every URL is cached"
3165        );
3166        assert!(
3167            warned(&over_cap, "cdp_url_cap"),
3168            "the cap denial must be attributable to the cap, not to some other condition"
3169        );
3170    }
3171
3172    #[test]
3173    fn cdp_cap_applies_in_fail_open_mode_too() {
3174        // The decision recorded in the rev-7 plan: this is a malformed-cert
3175        // rejection, not a revocation-unavailability denial, so opting out of
3176        // fail-closed does NOT opt out of the cap. An implementation that puts
3177        // the cap check inside the fail-closed branch admits here.
3178        let (over_cap, urls) =
3179            crl_set_with_cached_urls(fail_open_config(), MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE + 1);
3180        assert!(
3181            over_cap.__test_note_discovered_urls_by_cert(&urls, &[]),
3182            "the cap must deny in fail-open mode, where the same amplification is paid"
3183        );
3184
3185        let (at_cap, urls) =
3186            crl_set_with_cached_urls(fail_open_config(), MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE);
3187        assert!(
3188            !at_cap.__test_note_discovered_urls_by_cert(&urls, &[]),
3189            "the cap must not become a blanket fail-open denial"
3190        );
3191    }
3192
3193    #[test]
3194    fn cdp_cap_is_evaluated_before_out_of_band_mutation_detection() {
3195        let (set, urls) = crl_set_with_cached_urls(
3196            tamper_test_config(),
3197            MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE + 1,
3198        );
3199        let target = urls.first().expect("at least one url").clone();
3200        let committed = set
3201            .cache_lock()
3202            .try_read()
3203            .expect("uncontended")
3204            .get(&target)
3205            .cloned()
3206            .expect("entry present");
3207        set.cache_lock()
3208            .try_write()
3209            .expect("uncontended")
3210            .insert(target, same_shape_replacement(&committed));
3211
3212        assert!(set.__test_note_discovered_urls_by_cert(&urls, &[]));
3213        assert!(
3214            warned(&set, "cdp_url_cap"),
3215            "over-cap must be reported as the cap, so operators are not misdirected"
3216        );
3217        assert!(
3218            !warned(&set, "cache_entry_mismatch"),
3219            "the cap must short-circuit before any per-entry auditing runs"
3220        );
3221    }
3222
3223    // ---- B4-4 / B2: remaining precheck semantics --------------------------
3224
3225    #[tokio::test]
3226    async fn unrelated_commit_does_not_invalidate_other_urls() {
3227        // RELEASE-CRITICAL. `commit_cache_update_atomically` clones the live
3228        // cache, and cloning a `CachedCrl` reallocates its DER, so every
3229        // carried-forward entry gets a new address. An implementation that
3230        // carries identities forward instead of recomputing them denies every
3231        // handshake after any unrelated refresh.
3232        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
3233        let first = "https://first.example.test/crl".to_owned();
3234        let second = "https://second.example.test/crl".to_owned();
3235        let now = SystemTime::now();
3236
3237        set.__test_insert_cache(&first, synthetic_entry(now)).await;
3238        assert!(!set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&first), &[]));
3239
3240        set.__test_insert_cache(&second, synthetic_entry(now)).await;
3241
3242        assert!(
3243            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&first), &[]),
3244            "committing an unrelated URL must not invalidate an existing URL's identity"
3245        );
3246        assert!(
3247            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&second), &[]),
3248            "the newly committed URL must admit too"
3249        );
3250    }
3251
3252    #[tokio::test]
3253    async fn lock_contention_is_not_reported_as_out_of_band_mutation() {
3254        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
3255        let url = "https://contended.example.test/crl".to_owned();
3256        set.__test_insert_cache(&url, synthetic_entry(SystemTime::now()))
3257            .await;
3258
3259        let guard = set.cache_lock().write().await;
3260        assert!(
3261            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
3262            "contention must fall through to the committed state, not deny"
3263        );
3264        assert!(
3265            !warned(&set, "cache_entry_mismatch"),
3266            "contention must not be logged as out-of-band mutation, or operators chase a phantom"
3267        );
3268        drop(guard);
3269    }
3270
3271    #[tokio::test]
3272    async fn mutation_detection_only_ever_adds_a_denial() {
3273        let (set, _rx) = test_crl_set_with_receiver_config(tamper_test_config());
3274        let cached = "https://audited.example.test/crl".to_owned();
3275        let uncached = "https://never-cached.example.test/crl".to_owned();
3276        let now = SystemTime::now();
3277
3278        let committed = synthetic_entry(now);
3279        set.__test_insert_cache(&cached, committed.clone()).await;
3280        assert!(
3281            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&uncached), &[]),
3282            "an all-uncached certificate denies under all(not cached) before any mutation"
3283        );
3284
3285        set.__test_replace_cache_entry_unverified(&cached, same_shape_replacement(&committed))
3286            .await;
3287
3288        assert!(
3289            set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&uncached), &[]),
3290            "mutating an unrelated cached URL must not flip an all-uncached deny into an admit"
3291        );
3292    }
3293
3294    #[tokio::test]
3295    async fn fail_open_mode_still_admits_a_mutated_entry() {
3296        // Operators who set `crl_deny_on_unavailable = false` accepted that a
3297        // revoked cert is admitted when its CRL cannot be trusted. Silently
3298        // re-enabling denial for them would be a behaviour change they did not
3299        // opt into; only the malformed-cert cap applies in this mode.
3300        let (set, _rx) = test_crl_set_with_receiver_config(fail_open_config());
3301        let url = "https://fail-open.example.test/crl".to_owned();
3302        let committed = synthetic_entry(SystemTime::now());
3303
3304        set.__test_insert_cache(&url, committed.clone()).await;
3305        set.__test_replace_cache_entry_unverified(&url, same_shape_replacement(&committed))
3306            .await;
3307
3308        assert!(
3309            !set.__test_note_discovered_urls_by_cert(std::slice::from_ref(&url), &[]),
3310            "fail-open must stay fail-open for out-of-band mutation"
3311        );
3312    }
3313}