1use std::collections::HashMap;
33use std::sync::Arc;
34use std::time::Duration;
35
36use async_trait::async_trait;
37use time::OffsetDateTime;
38
39use crate::Clock;
40
41pub const DEFAULT_MAX_URL_LEN: usize = 2048;
44
45pub const DEFAULT_MAX_BATCH: usize = 500;
48
49pub const DEFAULT_CACHE_CAPACITY: usize = 4096;
51
52pub const DEFAULT_NEGATIVE_TTL: Duration = Duration::from_secs(300);
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
61#[non_exhaustive]
62pub enum ThreatType {
63 Malware,
65 SocialEngineering,
67 UnwantedSoftware,
70 PotentiallyHarmfulApplication,
72}
73
74impl ThreatType {
75 #[must_use]
78 pub const fn as_str(self) -> &'static str {
79 match self {
80 Self::Malware => "MALWARE",
81 Self::SocialEngineering => "SOCIAL_ENGINEERING",
82 Self::UnwantedSoftware => "UNWANTED_SOFTWARE",
83 Self::PotentiallyHarmfulApplication => "POTENTIALLY_HARMFUL_APPLICATION",
84 }
85 }
86}
87
88#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
90#[non_exhaustive]
91pub enum UrlSafetySource {
92 GoogleWebRisk,
94 GoogleSafeBrowsing,
96 Cache,
99}
100
101impl UrlSafetySource {
102 #[must_use]
104 pub const fn as_str(self) -> &'static str {
105 match self {
106 Self::GoogleWebRisk => "google-webrisk",
107 Self::GoogleSafeBrowsing => "google-safebrowsing",
108 Self::Cache => "cache",
109 }
110 }
111}
112
113#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct UrlVerdict {
117 pub url: String,
119 pub threats: Vec<ThreatType>,
121 pub source: UrlSafetySource,
123 pub checked_at: OffsetDateTime,
125 pub cache_ttl: Option<Duration>,
128 pub cached: bool,
130}
131
132impl UrlVerdict {
133 #[must_use]
135 pub fn is_threat(&self) -> bool {
136 !self.threats.is_empty()
137 }
138}
139
140#[derive(Debug, Clone, Copy, PartialEq, Eq)]
142#[non_exhaustive]
143pub enum UrlSafetyLimit {
144 UrlLength,
146 BatchSize,
148}
149
150impl UrlSafetyLimit {
151 #[must_use]
153 pub const fn name(self) -> &'static str {
154 match self {
155 Self::UrlLength => "url length",
156 Self::BatchSize => "batch size",
157 }
158 }
159}
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
164pub struct UrlSafetyLimits {
165 pub max_url_len: usize,
167 pub max_batch: usize,
169}
170
171impl Default for UrlSafetyLimits {
172 fn default() -> Self {
173 Self {
174 max_url_len: DEFAULT_MAX_URL_LEN,
175 max_batch: DEFAULT_MAX_BATCH,
176 }
177 }
178}
179
180#[derive(Debug, Clone, PartialEq, Eq)]
191#[non_exhaustive]
192pub enum UrlSafetyError {
193 InvalidUrl,
197 LimitExceeded {
199 limit: UrlSafetyLimit,
200 actual: usize,
201 max: usize,
202 },
203 Rejected { status: u16 },
206 Transient { retry_after: Option<Duration> },
209 Provider(String),
211}
212
213impl std::fmt::Display for UrlSafetyError {
214 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215 let scrub = crate::logging::scrub_text;
216 match self {
217 Self::InvalidUrl => f.write_str("not an http(s) URL; nothing was checked"),
218 Self::LimitExceeded { limit, actual, max } => {
219 write!(f, "over the {} ceiling: {actual} > {max}", limit.name())
220 }
221 Self::Rejected { status } => {
222 write!(
223 f,
224 "url safety provider rejected the request: status {status}"
225 )
226 }
227 Self::Transient { retry_after } => {
228 write!(
229 f,
230 "url safety call failed, retryable: retry after {retry_after:?}"
231 )
232 }
233 Self::Provider(message) => write!(f, "url safety provider failed: {}", scrub(message)),
234 }
235 }
236}
237
238impl std::error::Error for UrlSafetyError {}
239
240impl UrlSafetyError {
241 #[must_use]
243 pub fn retry_after(&self) -> Option<Duration> {
244 match self {
245 Self::Transient { retry_after } => *retry_after,
246 _ => None,
247 }
248 }
249}
250
251pub fn check_url(url: &str, limits: &UrlSafetyLimits) -> Result<(), UrlSafetyError> {
260 if url.len() > limits.max_url_len {
261 return Err(UrlSafetyError::LimitExceeded {
262 limit: UrlSafetyLimit::UrlLength,
263 actual: url.len(),
264 max: limits.max_url_len,
265 });
266 }
267 let parsed = url::Url::parse(url).map_err(|_| UrlSafetyError::InvalidUrl)?;
268 if parsed.scheme() != "http" && parsed.scheme() != "https" {
269 return Err(UrlSafetyError::InvalidUrl);
270 }
271 Ok(())
272}
273
274pub fn check_batch(urls: &[String], limits: &UrlSafetyLimits) -> Result<(), UrlSafetyError> {
284 if urls.len() > limits.max_batch {
285 return Err(UrlSafetyError::LimitExceeded {
286 limit: UrlSafetyLimit::BatchSize,
287 actual: urls.len(),
288 max: limits.max_batch,
289 });
290 }
291 for url in urls {
292 check_url(url, limits)?;
293 }
294 Ok(())
295}
296
297#[async_trait]
304pub trait UrlSafety: Send + Sync {
305 async fn check(&self, url: &str) -> Result<UrlVerdict, UrlSafetyError>;
313
314 async fn check_many(&self, urls: &[String]) -> Result<Vec<UrlVerdict>, UrlSafetyError> {
323 check_batch(urls, &UrlSafetyLimits::default())?;
324 let mut verdicts = Vec::with_capacity(urls.len());
325 for url in urls {
326 verdicts.push(self.check(url).await?);
327 }
328 Ok(verdicts)
329 }
330}
331
332#[allow(clippy::disallowed_types)]
338type Guarded<T> = std::sync::Mutex<T>;
339
340struct Entry {
342 verdict: UrlVerdict,
343 expires_at: OffsetDateTime,
344}
345
346impl Entry {
347 fn fresh_at(&self, now: OffsetDateTime) -> bool {
349 now < self.expires_at
350 }
351}
352
353pub struct CachedUrlSafety {
363 inner: Arc<dyn UrlSafety>,
364 clock: Arc<dyn Clock>,
365 #[allow(clippy::disallowed_types)]
366 state: Guarded<HashMap<String, Entry>>,
367 capacity: usize,
368 negative_ttl: Duration,
369}
370
371impl CachedUrlSafety {
372 #[must_use]
375 pub fn new(inner: Arc<dyn UrlSafety>, clock: Arc<dyn Clock>) -> Self {
376 Self {
377 inner,
378 clock,
379 state: Guarded::new(HashMap::new()),
380 capacity: DEFAULT_CACHE_CAPACITY,
381 negative_ttl: DEFAULT_NEGATIVE_TTL,
382 }
383 }
384
385 #[must_use]
393 pub fn with_capacity(mut self, capacity: usize) -> Self {
394 self.capacity = if capacity == 0 {
395 DEFAULT_CACHE_CAPACITY
396 } else {
397 capacity
398 };
399 self
400 }
401
402 #[must_use]
405 pub fn with_negative_ttl(mut self, ttl: Duration) -> Self {
406 self.negative_ttl = ttl;
407 self
408 }
409
410 fn fresh(&self, url: &str) -> Option<UrlVerdict> {
417 let now = self.clock.now();
418 let state = self.state.lock().expect("url safety cache");
419 state
420 .get(url)
421 .filter(|entry| entry.fresh_at(now))
422 .map(Self::served)
423 }
424
425 fn served(entry: &Entry) -> UrlVerdict {
428 let mut verdict = entry.verdict.clone();
429 verdict.cached = true;
430 verdict
431 }
432
433 fn store(&self, verdicts: &[UrlVerdict]) {
439 if verdicts.is_empty() {
440 return;
441 }
442 let now = self.clock.now();
443 let mut state = self.state.lock().expect("url safety cache");
444 for verdict in verdicts {
445 Self::insert(&mut state, verdict, now, self.capacity, self.negative_ttl);
446 }
447 }
448
449 fn insert(
453 state: &mut HashMap<String, Entry>,
454 verdict: &UrlVerdict,
455 now: OffsetDateTime,
456 capacity: usize,
457 negative_ttl: Duration,
458 ) {
459 state.retain(|_, entry| entry.fresh_at(now));
460 while state.len() >= capacity {
461 let Some(oldest) = state
462 .iter()
463 .min_by_key(|(_, entry)| entry.expires_at)
464 .map(|(url, _)| url.clone())
465 else {
466 break;
467 };
468 state.remove(&oldest);
469 }
470 let ttl = verdict.cache_ttl.unwrap_or(negative_ttl);
471 let expires_at = verdict
472 .checked_at
473 .saturating_add(time::Duration::try_from(ttl).unwrap_or(time::Duration::MAX));
474 state.insert(
475 verdict.url.clone(),
476 Entry {
477 verdict: verdict.clone(),
478 expires_at,
479 },
480 );
481 }
482
483 fn partition(&self, urls: &[String]) -> (Vec<Option<UrlVerdict>>, Vec<String>) {
491 let now = self.clock.now();
492 let state = self.state.lock().expect("url safety cache");
493 let mut slots = Vec::with_capacity(urls.len());
494 let mut misses = Vec::new();
495 for url in urls {
496 let hit = state
497 .get(url)
498 .filter(|entry| entry.fresh_at(now))
499 .map(Self::served);
500 if hit.is_none() {
501 misses.push(url.clone());
502 }
503 slots.push(hit);
504 }
505 drop(state);
506 (slots, misses)
507 }
508}
509
510#[async_trait]
511impl UrlSafety for CachedUrlSafety {
512 async fn check(&self, url: &str) -> Result<UrlVerdict, UrlSafetyError> {
513 if let Some(verdict) = self.fresh(url) {
514 return Ok(verdict);
515 }
516 let verdict = self.inner.check(url).await?;
518 self.store(std::slice::from_ref(&verdict));
519 Ok(verdict)
520 }
521
522 async fn check_many(&self, urls: &[String]) -> Result<Vec<UrlVerdict>, UrlSafetyError> {
523 check_batch(urls, &UrlSafetyLimits::default())?;
524 let (mut slots, misses) = self.partition(urls);
525 if !misses.is_empty() {
526 let fetched = self.inner.check_many(&misses).await?;
530 self.store(&fetched);
531 let mut fetched = fetched.into_iter();
532 for slot in &mut slots {
533 if slot.is_none() {
534 *slot = fetched.next();
535 }
536 }
537 }
538 Ok(slots.into_iter().flatten().collect())
542 }
543}
544
545impl std::fmt::Debug for CachedUrlSafety {
546 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
548 let entries = self
549 .state
550 .lock()
551 .map_or(0, |state: std::sync::MutexGuard<'_, _>| state.len());
552 f.debug_struct("CachedUrlSafety")
553 .field("capacity", &self.capacity)
554 .field("negative_ttl", &self.negative_ttl)
555 .field("entries", &entries)
556 .finish_non_exhaustive()
557 }
558}
559
560#[cfg(test)]
561mod tests {
562 use super::*;
563 use std::sync::atomic::{AtomicBool, AtomicI64, AtomicUsize, Ordering};
564
565 fn limits() -> UrlSafetyLimits {
566 UrlSafetyLimits {
567 max_url_len: 32,
568 max_batch: 2,
569 }
570 }
571
572 #[test]
573 fn threat_types_and_sources_name_themselves() {
574 assert_eq!(ThreatType::Malware.as_str(), "MALWARE");
575 assert_eq!(ThreatType::SocialEngineering.as_str(), "SOCIAL_ENGINEERING");
576 assert_eq!(ThreatType::UnwantedSoftware.as_str(), "UNWANTED_SOFTWARE");
577 assert_eq!(
578 ThreatType::PotentiallyHarmfulApplication.as_str(),
579 "POTENTIALLY_HARMFUL_APPLICATION"
580 );
581 assert_eq!(UrlSafetySource::GoogleWebRisk.as_str(), "google-webrisk");
582 assert_eq!(
583 UrlSafetySource::GoogleSafeBrowsing.as_str(),
584 "google-safebrowsing"
585 );
586 assert_eq!(UrlSafetySource::Cache.as_str(), "cache");
587 }
588
589 #[test]
590 fn check_url_accepts_http_and_https_and_refuses_the_rest() {
591 let limits = UrlSafetyLimits::default();
592 assert_eq!(check_url("https://example.com/", &limits), Ok(()));
593 assert_eq!(check_url("http://example.com/a?b=c", &limits), Ok(()));
594 assert_eq!(
595 check_url("ftp://example.com/file", &limits),
596 Err(UrlSafetyError::InvalidUrl)
597 );
598 assert_eq!(
600 check_url("example.com/path", &limits),
601 Err(UrlSafetyError::InvalidUrl)
602 );
603 assert_eq!(check_url("", &limits), Err(UrlSafetyError::InvalidUrl));
604 assert_eq!(
605 check_url("mailto:someone@example.com", &limits),
606 Err(UrlSafetyError::InvalidUrl)
607 );
608 }
609
610 #[test]
611 fn check_url_refuses_an_over_long_url_as_a_limit() {
612 let limits = limits();
613 let fits = format!("https://example.com/{}", "a".repeat(12));
616 assert_eq!(fits.len(), 32);
617 assert_eq!(check_url(&fits, &limits), Ok(()));
618 let long = format!("https://example.com/{}", "a".repeat(13));
619 assert_eq!(
620 check_url(&long, &limits),
621 Err(UrlSafetyError::LimitExceeded {
622 limit: UrlSafetyLimit::UrlLength,
623 actual: 33,
624 max: 32,
625 })
626 );
627 }
628
629 #[test]
630 fn check_batch_bounds_the_size_and_still_checks_each_url() {
631 let limits = limits();
632 assert_eq!(check_batch(&[], &limits), Ok(()), "an empty batch is fine");
633 let ok = vec![
634 "https://a.example".to_owned(),
635 "https://b.example".to_owned(),
636 ];
637 assert_eq!(check_batch(&ok, &limits), Ok(()));
638
639 let oversized = vec![
640 "https://a.example".to_owned(),
641 "https://b.example".to_owned(),
642 "https://c.example".to_owned(),
643 ];
644 assert_eq!(
645 check_batch(&oversized, &limits),
646 Err(UrlSafetyError::LimitExceeded {
647 limit: UrlSafetyLimit::BatchSize,
648 actual: 3,
649 max: 2,
650 })
651 );
652
653 let one_bad = vec!["https://a.example".to_owned(), "ftp://b.example".to_owned()];
655 assert_eq!(
656 check_batch(&one_bad, &limits),
657 Err(UrlSafetyError::InvalidUrl)
658 );
659 }
660
661 #[test]
662 fn provider_text_is_scrubbed_in_display_but_not_debug() {
663 let error =
664 UrlSafetyError::Provider("400 for https://api.example.test?token=secret".to_owned());
665 let text = error.to_string();
666 assert!(!text.contains("secret"), "{text}");
667 assert!(format!("{error:?}").contains("secret"));
668 assert_eq!(
671 UrlSafetyError::Rejected { status: 429 }.to_string(),
672 "url safety provider rejected the request: status 429"
673 );
674 assert_eq!(
675 UrlSafetyError::InvalidUrl.to_string(),
676 "not an http(s) URL; nothing was checked"
677 );
678 }
679
680 #[test]
681 fn transient_carries_an_optional_retry_after() {
682 let throttled = UrlSafetyError::Transient {
683 retry_after: Some(Duration::from_secs(30)),
684 };
685 assert_eq!(throttled.retry_after(), Some(Duration::from_secs(30)));
686 assert_eq!(
687 UrlSafetyError::Transient { retry_after: None }.retry_after(),
688 None
689 );
690 assert_eq!(UrlSafetyError::InvalidUrl.retry_after(), None);
691 }
692
693 struct FakeClock(AtomicI64);
695
696 impl FakeClock {
697 fn at(secs: i64) -> Arc<Self> {
698 Arc::new(Self(AtomicI64::new(secs)))
699 }
700
701 fn advance(&self, secs: i64) {
702 self.0.fetch_add(secs, Ordering::Relaxed);
703 }
704 }
705
706 impl Clock for FakeClock {
707 fn now(&self) -> OffsetDateTime {
708 OffsetDateTime::from_unix_timestamp(self.0.load(Ordering::Relaxed)).expect("in range")
709 }
710 }
711
712 struct FakeInner {
716 clock: Arc<FakeClock>,
717 checks: AtomicUsize,
718 batches: AtomicUsize,
719 fail: AtomicBool,
720 ttl: Option<Duration>,
723 }
724
725 impl FakeInner {
726 fn checks(&self) -> usize {
727 self.checks.load(Ordering::Relaxed)
728 }
729
730 fn batches(&self) -> usize {
731 self.batches.load(Ordering::Relaxed)
732 }
733
734 fn verdict(&self, url: &str) -> UrlVerdict {
735 let threats = if url.contains("malware") {
736 vec![ThreatType::Malware]
737 } else if url.contains("phishing") {
738 vec![ThreatType::SocialEngineering]
739 } else {
740 Vec::new()
741 };
742 UrlVerdict {
743 url: url.to_owned(),
744 threats,
745 source: UrlSafetySource::GoogleWebRisk,
746 checked_at: self.clock.now(),
747 cache_ttl: self.ttl,
748 cached: false,
749 }
750 }
751 }
752
753 #[async_trait]
754 impl UrlSafety for FakeInner {
755 async fn check(&self, url: &str) -> Result<UrlVerdict, UrlSafetyError> {
756 self.checks.fetch_add(1, Ordering::Relaxed);
757 if self.fail.load(Ordering::Relaxed) {
758 return Err(UrlSafetyError::Provider(format!("boom for {url}")));
759 }
760 Ok(self.verdict(url))
761 }
762
763 async fn check_many(&self, urls: &[String]) -> Result<Vec<UrlVerdict>, UrlSafetyError> {
764 self.batches.fetch_add(1, Ordering::Relaxed);
765 let mut verdicts = Vec::with_capacity(urls.len());
766 for url in urls {
767 if self.fail.load(Ordering::Relaxed) {
768 return Err(UrlSafetyError::Provider(format!("boom for {url}")));
769 }
770 verdicts.push(self.verdict(url));
771 }
772 Ok(verdicts)
773 }
774 }
775
776 type Parts = (Arc<FakeClock>, Arc<FakeInner>, CachedUrlSafety);
777
778 fn cache_with(ttl: Option<Duration>, fail: bool) -> Parts {
780 let clock = FakeClock::at(1_000);
781 let inner = Arc::new(FakeInner {
782 clock: clock.clone(),
783 checks: AtomicUsize::new(0),
784 batches: AtomicUsize::new(0),
785 fail: AtomicBool::new(fail),
786 ttl,
787 });
788 let cache = CachedUrlSafety::new(inner.clone(), clock.clone());
789 (clock, inner, cache)
790 }
791
792 #[test]
793 fn a_hit_within_the_ttl_serves_the_cache_and_a_miss_goes_to_the_provider() {
794 let (clock, inner, cache) = cache_with(Some(Duration::from_secs(60)), false);
795
796 let url = "https://example.test/page";
797 let first = pollster::block_on(cache.check(url)).expect("checked");
798 assert_eq!(inner.checks(), 1);
799 assert!(!first.cached);
800 assert_eq!(
801 first.url, url,
802 "the verdict's url equals the input, verbatim"
803 );
804 assert!(!first.is_threat());
805
806 clock.advance(59);
809 let second = pollster::block_on(cache.check(url)).expect("checked");
810 assert_eq!(inner.checks(), 1, "served from the cache");
811 assert!(second.cached);
812 assert_eq!(second.threats, first.threats);
813 assert_eq!(second.checked_at, first.checked_at);
814
815 clock.advance(2); pollster::block_on(cache.check(url)).expect("checked");
818 assert_eq!(inner.checks(), 2, "past the TTL the provider answers again");
819 }
820
821 #[test]
822 fn a_verdict_without_a_provider_ttl_uses_the_negative_ttl() {
823 let (clock, inner, cache) = cache_with(None, false);
824 let cache = cache.with_negative_ttl(Duration::from_secs(30));
825
826 pollster::block_on(cache.check("https://example.test")).expect("checked");
827 clock.advance(29);
828 pollster::block_on(cache.check("https://example.test")).expect("checked");
829 assert_eq!(inner.checks(), 1, "still inside the negative TTL");
830
831 clock.advance(2); pollster::block_on(cache.check("https://example.test")).expect("checked");
833 assert_eq!(inner.checks(), 2, "past the negative TTL");
834 }
835
836 #[test]
837 fn check_many_merges_in_input_order_with_one_batch_call_for_the_misses() {
838 let (_, inner, cache) = cache_with(Some(Duration::from_secs(300)), false);
839
840 pollster::block_on(cache.check("https://a.example")).expect("warmed");
843 assert_eq!(inner.checks(), 1);
844
845 let urls = vec![
846 "https://malware.example".to_owned(),
847 "https://a.example".to_owned(),
848 "https://phishing.example".to_owned(),
849 ];
850 let verdicts = pollster::block_on(cache.check_many(&urls)).expect("checked");
851 assert_eq!(verdicts.len(), urls.len(), "one verdict per input");
852 assert_eq!(inner.batches(), 1, "one batch call covered every miss");
853 assert_eq!(inner.checks(), 1, "the warm URL was not re-checked");
854 assert_eq!(verdicts[0].url, urls[0]);
855 assert_eq!(verdicts[0].threats, vec![ThreatType::Malware]);
856 assert!(!verdicts[0].cached);
857 assert_eq!(verdicts[1].url, urls[1], "the hit kept its input position");
858 assert!(verdicts[1].cached);
859 assert_eq!(verdicts[2].url, urls[2]);
860 assert_eq!(verdicts[2].threats, vec![ThreatType::SocialEngineering]);
861 assert!(!verdicts[2].cached);
862
863 let again = pollster::block_on(cache.check_many(&urls)).expect("checked");
865 assert_eq!(inner.batches(), 1, "nothing new reached the provider");
866 assert!(again.iter().all(|verdict| verdict.cached));
867 }
868
869 #[test]
870 fn the_cache_checks_limits_before_the_provider() {
871 let (_, inner, cache) = cache_with(Some(Duration::from_secs(60)), false);
872 let urls = (0..501)
873 .map(|n| format!("https://{n}.example"))
874 .collect::<Vec<_>>();
875 let error = pollster::block_on(cache.check_many(&urls)).expect_err("over the ceiling");
876 assert!(matches!(
877 error,
878 UrlSafetyError::LimitExceeded {
879 limit: UrlSafetyLimit::BatchSize,
880 ..
881 }
882 ));
883 assert_eq!(inner.batches(), 0, "the provider was never asked");
884 }
885
886 #[test]
887 fn a_full_cache_evicts_the_earliest_expiring_entry() {
888 let (_, inner, cache) = cache_with(Some(Duration::from_secs(60)), false);
889 let cache = cache.with_capacity(1);
890
891 pollster::block_on(cache.check("https://first.example")).expect("checked");
892 pollster::block_on(cache.check("https://second.example")).expect("checked");
893 assert_eq!(inner.checks(), 2);
894 pollster::block_on(cache.check("https://first.example")).expect("checked");
897 assert_eq!(inner.checks(), 3);
898 pollster::block_on(cache.check("https://second.example")).expect("checked");
901 assert_eq!(inner.checks(), 4);
902 }
903
904 #[test]
905 fn an_error_propagates_and_is_never_cached() {
906 let (_, inner, cache) = cache_with(Some(Duration::from_secs(60)), true);
907
908 let error = pollster::block_on(cache.check("https://example.test")).expect_err("failed");
909 assert!(matches!(error, UrlSafetyError::Provider(_)));
910 pollster::block_on(cache.check("https://example.test")).expect_err("failed again");
912 assert_eq!(inner.checks(), 2, "a failed check must not be cached");
913 }
914}