eric / rustwarden

//! Pluggable observer for federation runtime events (RFC 0028, RFC 0032).
//!
//! The federation layer emits domain- and actor-scoped events at the
//! well-defined chokepoints listed in [`FediverseEventObserver`]
//! (signature failure, replay, SSRF guard hit, inbox abuse, auth
//! challenge, invite action, delete-actor cascade). An external
//! observer — most typically the `rustwarden-fediverse-guard` crate —
//! receives those events and reacts (logs, counters, auto-ban,
//! cascade cleanup, etc.).
//!
//! The trait is defined here, not in `rustwarden-fediverse-guard`, to
//! keep the dependency direction unidirectional: `core ← fediverse ←
//! fediverse-guard`. Anyone can implement [`FediverseEventObserver`]
//! without depending on the guard crate.
//!
//! All hook methods carry **no return value**. Implementations must
//! treat the federation hot path as untouchable: failures inside the
//! observer (DB hiccup, network blip) must never propagate as request
//! errors. Use `tracing::warn!` or internal counters instead.
//!
//! # Default no-op
//!
//! Every method has a default empty body so a partial implementation
//! is fine — a consumer that only cares about signature failures
//! overrides `on_signature_failure` and leaves the rest as is.

use async_trait::async_trait;
use serde_json::Value;
use uuid::Uuid;

/// Pluggable observer contract. Implementations must be cheap to
/// share (`Arc`) and must never block the request path.
#[async_trait]
pub trait FediverseEventObserver: Send + Sync {
    /// A signed inbound request failed cryptographic verification.
    ///
    /// `domain` is the verified-but-rejected sender (e.g. extracted
    /// from `keyId`). `actor_uri` is the full AP id when available.
    /// `reason` is a short machine-stable string ("digest mismatch",
    /// "bad signature", "key not in cache", …).
    async fn on_signature_failure(&self, _domain: &str, _actor_uri: Option<&str>, _reason: &str) {}

    /// A replay was detected (stale `Date` header, duplicate nonce,
    /// or signature re-use). `details` is a free-form JSON object
    /// the detector can use to enrich logs.
    async fn on_replay_attempt(&self, _domain: &str, _actor_uri: Option<&str>, _details: Value) {}

    /// Local outbound fetcher refused to talk to a private / loopback
    /// / link-local IP (SSRF guard fired). `target_host` is the host
    /// part of the rejected URL.
    async fn on_fetch_ssrf(&self, _target_host: &str, _attempted_ip: Option<&str>) {}

    /// An inbox payload failed structural validation (body too large,
    /// schema rejected, unknown activity type, …).
    async fn on_inbox_abuse(&self, _domain: &str, _actor_uri: Option<&str>, _reason: &str) {}

    /// An AP-native auth challenge was initiated by a remote IdP.
    /// `handle` is the requested `@user@domain`.
    async fn on_auth_challenge(&self, _domain: &str, _handle: &str) {}

    /// An invite was accepted or rejected (`decision`: `"accept"` |
    /// `"reject"` | `"revoke"`).
    async fn on_invite_action(&self, _domain: &str, _decision: &str) {}

    /// A `Delete{Actor}` activity has been processed by
    /// [`crate::services::fediverse_signal_service::FediverseSignalService::apply_delete_actor_observed`]:
    /// the matching `remote_identities` row has been soft-deleted
    /// (its `disabled_at` is now `NOW()`). Consumers receive this
    /// hook so they can cascade the deletion onto locally-managed
    /// artifacts attached to the remote identity (SSH keys, remote
    /// PATs, team memberships, …).
    ///
    /// Invoked **after** the framework's own state mutation is
    /// committed, so consumer-side lookups by `remote_identity_id`
    /// observe the soft-deleted row. Idempotent semantics: the
    /// framework guarantees this hook is fired **at most once per
    /// actor** per inbound `Delete{Actor}` — if the row was already
    /// disabled before this delivery, the hook is **not** fired (the
    /// `rows_disabled` counter on the
    /// [`crate::services::fediverse_signal_service::DeleteOutcome`]
    /// returned by the service was 0).
    ///
    /// Errors must be swallowed by the implementation — the
    /// observer is best-effort by contract (RFC 0028).
    ///
    /// Arguments:
    /// - `domain` : lower-cased domain of the deleted actor (always
    ///   populated).
    /// - `ap_id` : full ActivityPub `id` of the deleted actor (e.g.
    ///   `https://a.tld/users/alice`).
    /// - `remote_identity_id` : surrogate UUID of the
    ///   `remote_identities` row that was just disabled. Consumers
    ///   key their cleanup on this id (it is the FK pointed at by
    ///   `fediverse_resource_shares.remote_identity_id`,
    ///   `team_members.remote_identity_id`, and any consumer-side
    ///   tables such as `ssh_keys.remote_identity_id`).
    async fn on_delete_actor(&self, _domain: &str, _ap_id: &str, _remote_identity_id: Uuid) {}
}

/// No-op observer. Wired as the default extension when no consumer
/// registers a real implementation. Permits the federation layer to
/// always emit unconditionally without an `Option`.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopFediverseObserver;

#[async_trait]
impl FediverseEventObserver for NoopFediverseObserver {}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct Counting {
        sig_calls: AtomicUsize,
        replay_calls: AtomicUsize,
    }

    #[async_trait]
    impl FediverseEventObserver for Counting {
        async fn on_signature_failure(
            &self,
            _domain: &str,
            _actor_uri: Option<&str>,
            _reason: &str,
        ) {
            self.sig_calls.fetch_add(1, Ordering::SeqCst);
        }

        async fn on_replay_attempt(
            &self,
            _domain: &str,
            _actor_uri: Option<&str>,
            _details: Value,
        ) {
            self.replay_calls.fetch_add(1, Ordering::SeqCst);
        }
        // other methods inherit the no-op default
    }

    #[tokio::test]
    async fn noop_observer_swallows_every_call() {
        let obs: Arc<dyn FediverseEventObserver> = Arc::new(NoopFediverseObserver);
        obs.on_signature_failure("x.tld", None, "bad sig").await;
        obs.on_replay_attempt("x.tld", None, serde_json::json!({}))
            .await;
        obs.on_fetch_ssrf("evil.local", Some("127.0.0.1")).await;
        obs.on_inbox_abuse("x.tld", None, "oversize").await;
        obs.on_auth_challenge("x.tld", "@alice@a.tld").await;
        obs.on_invite_action("x.tld", "accept").await;
        obs.on_delete_actor("x.tld", "https://x.tld/users/a", Uuid::new_v4())
            .await;
        // Nothing to assert; we only check that defaults compile and
        // do not panic when called through a trait object.
    }

    #[tokio::test]
    async fn partial_impl_uses_default_for_unspecified_hooks() {
        let counter = Arc::new(Counting {
            sig_calls: AtomicUsize::new(0),
            replay_calls: AtomicUsize::new(0),
        });
        let obs: Arc<dyn FediverseEventObserver> = counter.clone();
        obs.on_signature_failure("x.tld", None, "sig").await;
        obs.on_signature_failure("x.tld", None, "sig").await;
        obs.on_inbox_abuse("x.tld", None, "huge").await; // default no-op
        assert_eq!(counter.sig_calls.load(Ordering::SeqCst), 2);
        assert_eq!(counter.replay_calls.load(Ordering::SeqCst), 0);
    }
}