1use 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);
71const CRL_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
74const MAX_RELEVANT_CDP_URLS_PER_HANDSHAKE: usize = 64;
83
84#[derive(Clone, Debug)]
86#[non_exhaustive]
87pub struct CachedCrl {
88 pub der: CertificateRevocationListDer<'static>,
90 pub this_update: SystemTime,
92 pub next_update: Option<SystemTime>,
94 pub fetched_at: SystemTime,
96 pub source_url: String,
98}
99
100pub(crate) struct VerifierState {
110 verifier: Arc<dyn ClientCertVerifier>,
112 cached_urls: HashSet<String>,
114 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#[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
164fn entry_identity(entry: &CachedCrl) -> EntryIdentity {
167 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
205fn 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
215fn 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#[allow(
240 missing_debug_implementations,
241 reason = "contains ArcSwap and dyn verifier internals"
242)]
243#[non_exhaustive]
244pub struct CrlSet {
245 verifier_state: ArcSwap<VerifierState>,
248 commit_lock: tokio::sync::Mutex<()>,
258 #[deprecated(
269 since = "3.8.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 pub roots: Arc<RootCertStore>,
275 pub config: MtlsConfig,
277 pub discover_tx: mpsc::UnboundedSender<String>,
279 client: reqwest::Client,
280 seen_urls: Mutex<HashSet<String>>,
283 pending_urls: Mutex<HashSet<String>>,
293 global_fetch_sem: Arc<Semaphore>,
295 host_semaphores: Arc<tokio::sync::Mutex<HashMap<String, Arc<Semaphore>>>>,
300 discovery_limiter: Arc<DefaultDirectRateLimiter>,
311 max_response_bytes: u64,
314 last_cap_warn: Mutex<HashMap<&'static str, Instant>>,
315 #[cfg(any(test, feature = "test-helpers"))]
323 discovery_send_probe: Mutex<Option<DiscoverySendProbe>>,
324}
325
326#[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 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 .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 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 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 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 #[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 #[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 #[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 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 async fn commit_cache_update_atomically(
521 &self,
522 inserts: Vec<(String, CachedCrl)>,
523 removals: &[String],
524 ) -> Result<bool, RmcpServerKitError> {
525 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 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 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 drop(superseded);
585 }
586
587 {
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 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 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 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 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 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 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 #[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 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 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 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 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 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 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 let cache_guard = self.cache_lock().try_read();
894 let state = self.verifier_state.load_full();
895
896 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 let deny = relevant_urls
932 .iter()
933 .all(|url| !state.cached_urls.contains(url));
934 (deny, state)
935 }
936
937 #[doc(hidden)]
944 #[deprecated(
945 since = "3.8.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 #[doc(hidden)]
983 #[deprecated(
984 since = "3.8.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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 #[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 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 #[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 #[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
1340pub struct DynamicClientCertVerifier {
1343 inner: Arc<CrlSet>,
1344 dn_subjects: Vec<DistinguishedName>,
1345}
1346
1347impl DynamicClientCertVerifier {
1348 #[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 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#[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 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
1506fn 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#[allow(
1537 clippy::cognitive_complexity,
1538 reason = "bootstrap coordinates timeout, parallel fetches, and partial-cache recovery"
1539)]
1540pub 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 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 .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 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 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#[allow(
1679 clippy::cognitive_complexity,
1680 reason = "refresher loop intentionally handles shutdown, timer, and discovery in one select"
1681)]
1682pub 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 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
1722struct 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 Ok(true) => {
1764 pending_guard.disarm();
1765 pending_guard
1766 .set
1767 .promote_pending_to_seen(&pending_guard.url);
1768 }
1769 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
1789pub 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
1826pub 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
1872fn 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 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
1906async 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
1949async 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 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 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
2063const MAX_ASN1_TIMESTAMP_SECS: u64 = 253_402_300_799;
2067
2068fn 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 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 #[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 #[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 #[tokio::test]
2530 async fn fetch_crl_rejects_userinfo_without_echoing_credentials() {
2531 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 #[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 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 let year_1601 = asn1_time_to_system_time(asn1(-11_644_473_600));
2574 assert!(year_1601 <= UNIX_EPOCH);
2575
2576 assert!(asn1_time_to_system_time(asn1(-2)) <= UNIX_EPOCH);
2578
2579 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 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 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 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 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 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 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 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 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 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 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 #[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 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 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 #[tokio::test]
3226 async fn unrelated_commit_does_not_invalidate_other_urls() {
3227 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 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}