eric / rustwarden

//! Bridge between `rustwarden-fediverse` runtime events and the
//! guard's detection + ban pipeline.
//!
//! [`GuardObserver`] implements
//! [`rustwarden_fediverse::FediverseEventObserver`] and translates
//! each federation hook call into a [`GuardEvent`]:
//!
//! 1. Persist the event via [`crate::GuardStore::insert_event`].
//! 2. Optionally emit it through a [`GuardEventSink`] (file, OTel, …).
//! 3. Run every detector in the chain; a triggered detector calls
//!    [`crate::BanManager::auto_ban`] which inserts the ban row.
//!
//! Hot-path safety: every step swallows errors via `tracing::warn!`
//! rather than propagating, so a downstream hiccup never breaks the
//! federation request path.

use std::sync::Arc;

use async_trait::async_trait;
use rustwarden_fediverse::observer::FediverseEventObserver;
use serde_json::Value;

use crate::ban::BanManager;
use crate::detector::{Detector, run_all};
use crate::events::{GuardEvent, GuardEventKind, GuardEventSink, Severity};
use crate::store::normalise_domain;

/// Connect federation events to the guard pipeline.
pub struct GuardObserver {
    manager: BanManager,
    detectors: Vec<Arc<dyn Detector>>,
    sink: Option<Arc<dyn GuardEventSink>>,
}

impl GuardObserver {
    /// Build a new observer wrapping the supplied ban manager and
    /// detector chain. Pass `None` for `sink` when events only need
    /// to land in the store (the typical production setup).
    #[must_use]
    pub fn new(
        manager: BanManager,
        detectors: Vec<Arc<dyn Detector>>,
        sink: Option<Arc<dyn GuardEventSink>>,
    ) -> Self {
        Self {
            manager,
            detectors,
            sink,
        }
    }

    /// Borrow the underlying [`BanManager`] (useful for admin
    /// endpoints that read or mutate state directly).
    #[must_use]
    pub fn manager(&self) -> &BanManager {
        &self.manager
    }

    async fn record(
        &self,
        kind: GuardEventKind,
        domain: &str,
        actor_uri: Option<&str>,
        severity: Severity,
        details: Value,
    ) {
        if !self.manager.config().enabled {
            return;
        }
        let domain = normalise_domain(domain);
        if domain.is_empty() {
            return;
        }
        let mut evt = GuardEvent::new(kind, &domain, severity);
        evt.actor_uri = actor_uri.map(str::to_string);
        evt.details = details;
        if let Err(e) = self.manager.store().insert_event(&evt).await {
            tracing::warn!(error = %e, "guard observer: failed to persist event");
            return;
        }
        if let Some(sink) = &self.sink {
            sink.emit(&evt).await;
        }
        if let Err(e) = run_all(&self.detectors, &self.manager, &domain).await {
            tracing::warn!(error = %e, "guard observer: detector chain failed");
        }
    }
}

#[async_trait]
impl FediverseEventObserver for GuardObserver {
    async fn on_signature_failure(&self, domain: &str, actor_uri: Option<&str>, reason: &str) {
        self.record(
            GuardEventKind::SignatureFailure,
            domain,
            actor_uri,
            Severity::High,
            serde_json::json!({"reason": reason}),
        )
        .await;
    }

    async fn on_replay_attempt(&self, domain: &str, actor_uri: Option<&str>, details: Value) {
        self.record(
            GuardEventKind::ReplayAttempt,
            domain,
            actor_uri,
            Severity::High,
            details,
        )
        .await;
    }

    async fn on_fetch_ssrf(&self, target_host: &str, attempted_ip: Option<&str>) {
        let details = serde_json::json!({"attempted_ip": attempted_ip});
        self.record(
            GuardEventKind::FetchSsrf,
            target_host,
            None,
            Severity::Critical,
            details,
        )
        .await;
    }

    async fn on_inbox_abuse(&self, domain: &str, actor_uri: Option<&str>, reason: &str) {
        self.record(
            GuardEventKind::InboxAbuse,
            domain,
            actor_uri,
            Severity::Medium,
            serde_json::json!({"reason": reason}),
        )
        .await;
    }

    async fn on_auth_challenge(&self, domain: &str, handle: &str) {
        self.record(
            GuardEventKind::AuthSpam,
            domain,
            None,
            Severity::Low,
            serde_json::json!({"handle": handle}),
        )
        .await;
    }

    async fn on_invite_action(&self, _domain: &str, _decision: &str) {
        // Invites are not a security event by themselves and already
        // emit an `admin_id` tracing log from `invite_routes.rs`
        // (ANSSI FUI-AUTH-1). Skip to avoid polluting the guard with
        // benign traffic. A future detector specific to invite churn
        // can reuse this hook.
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::GuardConfig;
    use crate::detector::default_detectors;
    use crate::events::GuardEventKind;
    use crate::store::MemoryStore;
    use chrono::Utc;

    fn observer() -> GuardObserver {
        let cfg = GuardConfig::for_tests();
        let store: Arc<dyn crate::GuardStore> = Arc::new(MemoryStore::new());
        let manager = BanManager::new(store, cfg.clone());
        let detectors = default_detectors(&cfg);
        GuardObserver::new(manager, detectors, None)
    }

    #[tokio::test]
    async fn on_signature_failure_persists_event() {
        let obs = observer();
        obs.on_signature_failure("evil.tld", Some("https://evil.tld/users/x"), "bad sig")
            .await;
        let since = Utc::now() - chrono::Duration::seconds(60);
        let count = obs
            .manager()
            .store()
            .count_events_since("evil.tld", GuardEventKind::SignatureFailure, since)
            .await
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn repeated_signature_failures_trigger_auto_ban() {
        let obs = observer();
        // for_tests threshold is 3 → 4 failures must trip the ban.
        for _ in 0..4 {
            obs.on_signature_failure("evil.tld", None, "bad sig").await;
        }
        let active = obs
            .manager()
            .store()
            .active_bans_for_domain("evil.tld", Utc::now())
            .await
            .unwrap();
        assert_eq!(active.len(), 1);
        assert!(active[0].triggered_by.starts_with("auto:signature_failure"));
    }

    #[tokio::test]
    async fn disabled_guard_swallows_calls_without_storing() {
        let mut cfg = GuardConfig::for_tests();
        cfg.enabled = false;
        let store: Arc<dyn crate::GuardStore> = Arc::new(MemoryStore::new());
        let manager = BanManager::new(store, cfg.clone());
        let detectors = default_detectors(&cfg);
        let obs = GuardObserver::new(manager, detectors, None);
        obs.on_signature_failure("evil.tld", None, "x").await;
        let since = Utc::now() - chrono::Duration::seconds(60);
        let count = obs
            .manager()
            .store()
            .count_events_since("evil.tld", GuardEventKind::SignatureFailure, since)
            .await
            .unwrap();
        assert_eq!(count, 0);
    }

    #[tokio::test]
    async fn normalises_domain_before_recording() {
        let obs = observer();
        obs.on_signature_failure("  HTTPS://Evil.TLD/  ", None, "x")
            .await;
        let since = Utc::now() - chrono::Duration::seconds(60);
        // The event must be stored under the canonical lower-cased form.
        let count = obs
            .manager()
            .store()
            .count_events_since("evil.tld", GuardEventKind::SignatureFailure, since)
            .await
            .unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn invite_action_is_no_op() {
        let obs = observer();
        obs.on_invite_action("a.tld", "accept").await;
        let since = Utc::now() - chrono::Duration::seconds(60);
        for kind in [
            GuardEventKind::AuthSpam,
            GuardEventKind::InboxAbuse,
            GuardEventKind::SignatureFailure,
        ] {
            let count = obs
                .manager()
                .store()
                .count_events_since("a.tld", kind, since)
                .await
                .unwrap();
            assert_eq!(count, 0, "kind {kind:?} should not be recorded");
        }
    }
}