eric / rustwarden

//! Sliding-window detector chain for the fediverse guard.
//!
//! Each detector watches one [`GuardEventKind`] and triggers an
//! auto-ban via [`BanManager`] when the count of qualifying events
//! within the sliding window strictly exceeds the configured
//! threshold. Detectors are stateless: they query the
//! [`crate::store::GuardStore`] for the count at evaluation time, so
//! they tolerate process restarts and concurrent producers without
//! coordination.
//!
//! The five canonical detectors map 1-to-1 to the five
//! [`GuardEventKind`] variants emitted by the federation layer:
//!
//! | Detector            | Source event             |
//! |---------------------|--------------------------|
//! | `signature_failure` | `SignatureFailure`       |
//! | `replay_attempt`    | `ReplayAttempt`          |
//! | `fetch_ssrf`        | `FetchSsrf`              |
//! | `inbox_abuse`       | `InboxAbuse`             |
//! | `auth_spam`         | `AuthSpam`               |
//!
//! Build the canonical set with [`default_detectors`] and run them
//! against a fresh event's domain with [`run_all`].

use std::sync::Arc;

use async_trait::async_trait;
use chrono::Utc;
use rustwarden_core::error::AppError;

use crate::ban::BanManager;
use crate::config::{DetectorThreshold, GuardConfig};
use crate::events::GuardEventKind;

/// Sliding-window detector contract.
#[async_trait]
pub trait Detector: Send + Sync {
    /// Static identifier captured in `BanEntry.triggered_by`
    /// as `"auto:<name>"`. Must be stable across releases — admins
    /// will see it in audit logs.
    fn name(&self) -> &'static str;

    /// Event kind this detector counts.
    fn kind(&self) -> GuardEventKind;

    /// Inspect the rolling window for `domain` and return `true` if
    /// the threshold was exceeded *and* a new ban was inserted.
    /// Returns `false` when the count is below threshold, when an
    /// active ban already exists, when the guard is in dry-run, or
    /// when the domain is on the allow ACL.
    async fn check(&self, manager: &BanManager, domain: &str) -> Result<bool, AppError>;
}

/// Generic threshold-based detector. Five instances cover the
/// canonical chain (see [`default_detectors`]).
#[derive(Clone)]
pub struct ThresholdDetector {
    name: &'static str,
    kind: GuardEventKind,
    threshold: DetectorThreshold,
}

impl ThresholdDetector {
    /// Build a detector watching `kind` against `threshold`.
    #[must_use]
    pub const fn new(
        name: &'static str,
        kind: GuardEventKind,
        threshold: DetectorThreshold,
    ) -> Self {
        Self {
            name,
            kind,
            threshold,
        }
    }

    /// Returns the threshold (count + window) carried by this detector.
    #[must_use]
    pub fn threshold(&self) -> DetectorThreshold {
        self.threshold
    }
}

#[async_trait]
impl Detector for ThresholdDetector {
    fn name(&self) -> &'static str {
        self.name
    }

    fn kind(&self) -> GuardEventKind {
        self.kind
    }

    async fn check(&self, manager: &BanManager, domain: &str) -> Result<bool, AppError> {
        if !manager.config().enabled {
            return Ok(false);
        }
        let window = chrono::Duration::from_std(self.threshold.window)
            .unwrap_or_else(|_| chrono::Duration::seconds(300));
        let since = Utc::now() - window;
        let count = manager
            .store()
            .count_events_since(domain, self.kind, since)
            .await?;
        if count > u64::from(self.threshold.count) {
            let reason = format!(
                "threshold exceeded: {count} events of {kind:?} within {window_secs}s",
                kind = self.kind,
                window_secs = self.threshold.window.as_secs(),
            );
            let ban = manager.auto_ban(domain, self.name, reason).await?;
            return Ok(ban.is_some());
        }
        Ok(false)
    }
}

/// Build the canonical detector chain aligned with the supplied
/// [`GuardConfig`].
#[must_use]
pub fn default_detectors(config: &GuardConfig) -> Vec<Arc<dyn Detector>> {
    vec![
        Arc::new(ThresholdDetector::new(
            "signature_failure",
            GuardEventKind::SignatureFailure,
            config.signature_failures,
        )),
        Arc::new(ThresholdDetector::new(
            "replay_attempt",
            GuardEventKind::ReplayAttempt,
            config.replay_attempts,
        )),
        Arc::new(ThresholdDetector::new(
            "fetch_ssrf",
            GuardEventKind::FetchSsrf,
            config.fetch_ssrf,
        )),
        Arc::new(ThresholdDetector::new(
            "inbox_abuse",
            GuardEventKind::InboxAbuse,
            config.inbox_abuse,
        )),
        Arc::new(ThresholdDetector::new(
            "auth_spam",
            GuardEventKind::AuthSpam,
            config.auth_spam,
        )),
    ]
}

/// Run every detector against `domain` and return the names of those
/// that triggered a ban this call. In practice the list contains 0
/// or 1 entries because once `BanManager::auto_ban` inserts a ban,
/// any subsequent detector finds the existing active row and skips.
///
/// # Errors
///
/// Propagates errors from the underlying store.
pub async fn run_all(
    detectors: &[Arc<dyn Detector>],
    manager: &BanManager,
    domain: &str,
) -> Result<Vec<&'static str>, AppError> {
    let mut triggered = Vec::new();
    for det in detectors {
        if det.check(manager, domain).await? {
            triggered.push(det.name());
        }
    }
    Ok(triggered)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::events::{GuardEvent, Severity};
    use crate::store::{AclEntry, AclKind, GuardStore, MemoryStore};
    use uuid::Uuid;

    fn manager() -> BanManager {
        BanManager::new(Arc::new(MemoryStore::new()), GuardConfig::for_tests())
    }

    fn manager_with(cfg: GuardConfig) -> BanManager {
        BanManager::new(Arc::new(MemoryStore::new()), cfg)
    }

    async fn push_events(store: &Arc<dyn GuardStore>, kind: GuardEventKind, domain: &str, n: u32) {
        for _ in 0..n {
            let evt = GuardEvent::new(kind, domain, Severity::Medium);
            store.insert_event(&evt).await.unwrap();
        }
    }

    fn signature_detector() -> ThresholdDetector {
        ThresholdDetector::new(
            "signature_failure",
            GuardEventKind::SignatureFailure,
            DetectorThreshold::new(3, 60),
        )
    }

    #[tokio::test]
    async fn check_does_not_trigger_at_threshold_boundary() {
        let m = manager();
        let det = signature_detector();
        // Threshold 3 means "strictly more than 3" → 3 events do not trigger.
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 3).await;
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn check_triggers_when_threshold_strictly_exceeded() {
        let m = manager();
        let det = signature_detector();
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 4).await;
        assert!(det.check(&m, "x.tld").await.unwrap());
        // Ban actually persisted.
        let active = m
            .store()
            .active_bans_for_domain("x.tld", Utc::now())
            .await
            .unwrap();
        assert_eq!(active.len(), 1);
        assert!(active[0].triggered_by.starts_with("auto:signature_failure"));
    }

    #[tokio::test]
    async fn check_is_idempotent_once_banned() {
        let m = manager();
        let det = signature_detector();
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 10).await;
        assert!(det.check(&m, "x.tld").await.unwrap());
        // Second run should not insert a new ban (already active).
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn check_ignores_other_event_kinds() {
        let m = manager();
        let det = signature_detector();
        // Inject 10 ReplayAttempt events — irrelevant for sig detector.
        push_events(m.store(), GuardEventKind::ReplayAttempt, "x.tld", 10).await;
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn allow_acl_prevents_trigger() {
        let m = manager();
        m.upsert_acl(&AclEntry {
            id: Uuid::new_v4(),
            kind: AclKind::Allow,
            domain: "trusted.tld".into(),
            reason: None,
            created_by: Uuid::new_v4(),
            created_at: Utc::now(),
            updated_at: Utc::now(),
        })
        .await
        .unwrap();
        let det = signature_detector();
        push_events(
            m.store(),
            GuardEventKind::SignatureFailure,
            "trusted.tld",
            10,
        )
        .await;
        assert!(!det.check(&m, "trusted.tld").await.unwrap());
    }

    #[tokio::test]
    async fn dry_run_inhibits_ban_insertion() {
        let mut cfg = GuardConfig::for_tests();
        cfg.dry_run = true;
        let m = manager_with(cfg);
        let det = signature_detector();
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 10).await;
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn disabled_guard_short_circuits_check() {
        let mut cfg = GuardConfig::for_tests();
        cfg.enabled = false;
        let m = manager_with(cfg);
        let det = signature_detector();
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 10).await;
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn sliding_window_excludes_old_events() {
        let m = manager();
        // Threshold 3 within 60s. Inject 4 events but all dated 10 min
        // in the past — the detector must see zero in the current window.
        let store = m.store().clone();
        let old = Utc::now() - chrono::Duration::minutes(10);
        for _ in 0..4 {
            let mut evt =
                GuardEvent::new(GuardEventKind::SignatureFailure, "x.tld", Severity::Medium);
            evt.created_at = old;
            store.insert_event(&evt).await.unwrap();
        }
        let det = signature_detector();
        assert!(!det.check(&m, "x.tld").await.unwrap());
    }

    #[tokio::test]
    async fn default_detectors_covers_five_kinds() {
        let cfg = GuardConfig::for_tests();
        let dets = default_detectors(&cfg);
        assert_eq!(dets.len(), 5);
        let kinds: Vec<GuardEventKind> = dets.iter().map(|d| d.kind()).collect();
        assert!(kinds.contains(&GuardEventKind::SignatureFailure));
        assert!(kinds.contains(&GuardEventKind::ReplayAttempt));
        assert!(kinds.contains(&GuardEventKind::FetchSsrf));
        assert!(kinds.contains(&GuardEventKind::InboxAbuse));
        assert!(kinds.contains(&GuardEventKind::AuthSpam));
    }

    #[tokio::test]
    async fn run_all_returns_triggered_names_only() {
        let m = manager();
        let cfg = GuardConfig::for_tests();
        let dets = default_detectors(&cfg);
        // Push enough signature failures to trip the first detector.
        push_events(m.store(), GuardEventKind::SignatureFailure, "x.tld", 10).await;
        let fired = run_all(&dets, &m, "x.tld").await.unwrap();
        assert_eq!(fired, vec!["signature_failure"]);
    }
}