GitRust
fediverse_signal_service.rs
1428 lignes · 54543 octets
//! Fediverse lifecycle signal handlers (RFC 0010). //! //! Applies incoming AP `Update{Actor}` / `Delete{Actor}` / //! `Revoke{Membership}` signals to the local state. The HTTP-Sig //! verification of the request that carried the signal is delegated to //! `SignatureVerifier` — the implementation that ships with this RFC //! is `AlwaysAcceptVerifier` (test-only); a follow-up RFC will plug //! the real RFC 9421 verifier. //! use std::sync::Arc; use async_trait::async_trait; use chrono::Utc; use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set}; use uuid::Uuid; use crate::models::remote_identity; use crate::observer::{FediverseEventObserver, NoopFediverseObserver}; use rustwarden_core::error::AppError; use rustwarden_core::models::team_member; /// Carries the actor metadata extracted from an inbound `Update{Actor}` /// activity. #[derive(Debug, Clone)] pub struct UpdatedActor { /// Canonical AP `id`. Match key against `remote_identities.ap_id`. pub ap_id: String, /// Optional display name (`name` field on the AP `Person`). pub display_name: Option<String>, /// Optional bio (`summary` field). pub summary_html: Option<String>, /// Optional avatar URL (`icon.url`). pub icon_url: Option<String>, /// Optional refreshed public key. pub public_key_pem: Option<String>, /// Optional raw activity body for audit / re-processing. pub raw_json: Option<serde_json::Value>, } /// Counters returned by [`FediverseSignalService::apply_delete_actor_with_hooks`]. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct DeleteOutcome { /// Number of `remote_identities` rows soft-deleted. pub rows_disabled: u64, /// Number of JWTs blacklisted (always `0` when no /// [`DeleteActorHooks::iterator`] is supplied). pub jwts_blacklisted: u64, } /// Pluggable side-effects for a remote-actor deletion (RFC 0033). /// /// All fields are `Option` with `None` as default (no-op). The /// recommended construction idiom is the struct-literal pattern with /// `..Default::default()`, which keeps existing call-sites compiling /// when new hooks get added in future RFCs: /// /// ```rust,ignore /// use rustwarden_fediverse::services::fediverse_signal_service::DeleteActorHooks; /// let hooks = DeleteActorHooks { /// observer: Some(&my_observer), /// ..Default::default() /// }; /// ``` /// /// The struct is **not** `#[non_exhaustive]` so consumers can keep /// instantiating it with struct literals; rely on /// `..Default::default()` for forward-compatibility instead. #[derive(Default)] pub struct DeleteActorHooks<'a> { /// Best-effort observer notified **after** the `disabled_at` /// flip has been committed (RFC 0032). `None` ⇒ no hook fires. pub observer: Option<&'a dyn FediverseEventObserver>, /// Enumerator of the active JWTs to blacklist for the deleted /// remote identity (RFC 0018). `None` ⇒ `jwts_blacklisted = 0`. pub iterator: Option<&'a dyn rustwarden_core::services::jwt_blacklist_service::ActiveTokenIterator>, } /// Stateless service that applies AP lifecycle signals. pub struct FediverseSignalService; impl FediverseSignalService { /// Apply an inbound `Update{Actor}`: upsert the row keyed by /// `ap_id`. The local fields not carried by `actor` are preserved /// (no destructive overwrite). pub async fn apply_update_actor( db: &DatabaseConnection, actor: UpdatedActor, ) -> Result<remote_identity::Model, AppError> { let existing = remote_identity::Entity::find() .filter(remote_identity::Column::ApId.eq(actor.ap_id.clone())) .one(db) .await?; let now = Utc::now(); if let Some(model) = existing { let mut active: remote_identity::ActiveModel = model.into(); if let Some(name) = actor.display_name { active.display_name = Set(Some(name)); } if let Some(summary) = actor.summary_html { active.summary_html = Set(Some(summary)); } if let Some(icon) = actor.icon_url { active.icon_url = Set(Some(icon)); } if let Some(pem) = actor.public_key_pem { active.public_key_pem = Set(pem); } if let Some(raw) = actor.raw_json { active.raw_json = Set(Some(raw)); } active.last_fetched_at = Set(now); Ok(active.update(db).await?) } else { // Minimal insert. The caller (inbox handler) is responsible // for filling in the rest via webfinger lookup before // calling apply_update_actor — but if the row truly is new // we still want to persist *something* rather than 404. let username_domain = parse_handle_from_ap_id(&actor.ap_id) .unwrap_or_else(|| ("unknown".to_string(), "unknown".to_string())); let active = remote_identity::ActiveModel { id: Set(Uuid::new_v4()), ap_id: Set(actor.ap_id.clone()), username: Set(username_domain.0), domain: Set(username_domain.1), display_name: Set(actor.display_name), summary_html: Set(actor.summary_html), inbox_url: Set(format!("{}/inbox", actor.ap_id)), shared_inbox_url: Set(None), public_key_pem: Set(actor.public_key_pem.unwrap_or_default()), icon_url: Set(actor.icon_url), actor_type: Set("Person".to_string()), raw_json: Set(actor.raw_json), last_fetched_at: Set(now), disabled_at: Set(None), created_at: Set(now), updated_at: Set(now), }; Ok(active.insert(db).await?) } } /// Canonical entrypoint for an inbound `Delete{Actor}` (RFC 0033). /// /// Combines the semantics of every historical variant: /// /// 1. Flip `remote_identities.disabled_at` to `NOW()` /// (idempotent: a re-delivery does nothing). /// 2. If [`DeleteActorHooks::iterator`] is `Some`, blacklist every /// active JWT enumerated for the deleted `remote_identity_id` /// (RFC 0018). /// 3. If [`DeleteActorHooks::observer`] is `Some`, fire /// [`FediverseEventObserver::on_delete_actor`] **after** the /// state mutation is committed (RFC 0032). /// /// Steps 2 and 3 only run on a real transition (active → /// disabled). Re-deliveries and unknown actors are silent. pub async fn apply_delete_actor_with_hooks( db: &DatabaseConnection, ap_id: &str, hooks: DeleteActorHooks<'_>, ) -> Result<DeleteOutcome, AppError> { let Some(row) = remote_identity::Entity::find() .filter(remote_identity::Column::ApId.eq(ap_id)) .one(db) .await? else { return Ok(DeleteOutcome::default()); }; if row.disabled_at.is_some() { return Ok(DeleteOutcome { rows_disabled: 0, jwts_blacklisted: 0, }); } let domain = row.domain.clone(); let remote_identity_id = row.id; let mut active: remote_identity::ActiveModel = row.into(); active.disabled_at = Set(Some(Utc::now())); active.update(db).await?; let jwts_blacklisted = if let Some(it) = hooks.iterator { rustwarden_core::services::jwt_blacklist_service::JwtBlacklistService::blacklist_by_remote_identity_id( db, remote_identity_id, it, ) .await? } else { 0 }; if let Some(observer) = hooks.observer { observer .on_delete_actor(&domain, ap_id, remote_identity_id) .await; } Ok(DeleteOutcome { rows_disabled: 1, jwts_blacklisted, }) } /// Compatibility wrapper. Prefer /// [`Self::apply_delete_actor_with_hooks`] for new code. pub async fn apply_delete_actor( db: &DatabaseConnection, ap_id: &str, ) -> Result<DeleteOutcome, AppError> { Self::apply_delete_actor_with_hooks(db, ap_id, DeleteActorHooks::default()).await } /// Compatibility wrapper. Prefer /// [`Self::apply_delete_actor_with_hooks`] for new code. pub async fn apply_delete_actor_observed( db: &DatabaseConnection, ap_id: &str, observer: &dyn FediverseEventObserver, ) -> Result<DeleteOutcome, AppError> { Self::apply_delete_actor_with_hooks( db, ap_id, DeleteActorHooks { observer: Some(observer), ..Default::default() }, ) .await } /// Compatibility wrapper. Prefer /// [`Self::apply_delete_actor_with_hooks`] for new code. pub async fn apply_delete_actor_with_iterator( db: &DatabaseConnection, ap_id: &str, iterator: &dyn rustwarden_core::services::jwt_blacklist_service::ActiveTokenIterator, ) -> Result<DeleteOutcome, AppError> { Self::apply_delete_actor_with_hooks( db, ap_id, DeleteActorHooks { iterator: Some(iterator), ..Default::default() }, ) .await } /// Apply an inbound `Revoke{Membership}`: drop the matching /// `team_members` row. Returns `true` if a row was deleted. pub async fn apply_revoke_membership( db: &DatabaseConnection, team_id: Uuid, remote_identity_id: Uuid, ) -> Result<bool, AppError> { let result = team_member::Entity::delete_many() .filter(team_member::Column::TeamId.eq(team_id)) .filter(team_member::Column::RemoteIdentityId.eq(Some(remote_identity_id))) .exec(db) .await?; Ok(result.rows_affected > 0) } } /// Inbound request context passed to [`SignatureVerifier::verify`] /// (RFC 0016). Carries every field needed by draft-cavage HTTP /// signature verification: method + path + headers + body + the /// actor's resolved public key. #[derive(Debug)] pub struct VerifyCtx<'a> { /// AP id of the signer (extracted from the body by the caller). pub actor_ap_id: &'a str, /// HTTP method, lowercased (`"post"`). pub method: &'a str, /// Path-and-query (`"/inbox"`). pub path: &'a str, /// All inbound headers (case-insensitive lookup at the caller). pub headers: &'a axum::http::HeaderMap, /// Raw body bytes (used by the caller for digest / audit). pub body: &'a [u8], /// Resolved PEM public key for `actor_ap_id`. Empty when the /// caller has not yet resolved the actor; impls may either /// reject or fetch on their own. pub public_key_pem: &'a str, } /// HTTP-Sig verifier for inbound AP activities. /// /// Implementations decide whether to accept a given signature. The /// `AlwaysAcceptVerifier` shipped here is **test/dev only**. /// `HttpSigVerifier` from /// [`crate::http_signatures`] is the production /// implementation. #[async_trait] pub trait SignatureVerifier: Send + Sync { /// Verify a signature given the full request context. async fn verify(&self, ctx: &VerifyCtx<'_>) -> Result<(), AppError>; } /// Test/dev verifier that accepts any signature. /// /// **Never use in production.** A logged warning emphasises the /// hazard at instantiation in production builds. /// /// When constructed with [`AlwaysAcceptVerifier::with_resolver`], the /// verifier also fetches the actor over HTTP and upserts the /// `remote_identities` row before returning `Ok(())` — this preserves /// the contract documented by [`crate::services::fediverse_auth_service::FediverseAuthService::persist_inbound_request`] /// (« the signature verifier ran upstream and already required this /// row ») so downstream services don't need to special-case the debug /// verifier. pub struct AlwaysAcceptVerifier { resolver: Option<(DatabaseConnection, reqwest::Client)>, observer: Arc<dyn FediverseEventObserver>, } impl AlwaysAcceptVerifier { /// Verifier with no side-effects. Suitable for unit tests where the /// caller pre-seeds `remote_identities` themselves. pub fn new() -> Self { Self { resolver: None, observer: Arc::new(NoopFediverseObserver), } } /// Verifier that, on each accepted request, fetches the actor doc /// from `ctx.actor_ap_id` over HTTP and upserts the corresponding /// `remote_identities` cache row. Best-effort: a fetch/parse failure /// is logged and the request is still accepted (since this verifier /// is « always accept » by contract). /// /// The internal HTTP client has [`ACTOR_FETCH_TIMEOUT`] applied. The /// SSRF guard inside [`fetch_and_cache_remote_actor`] is **bypassed** /// here (`allow_private_ips=true`) because this verifier is test/dev /// only and the docker-compose testbed lives on private IPs. pub fn with_resolver(db: DatabaseConnection) -> Self { let http = reqwest::Client::builder() .timeout(ACTOR_FETCH_TIMEOUT) .build() .expect("reqwest client builder must not fail"); Self { resolver: Some((db, http)), observer: Arc::new(NoopFediverseObserver), } } /// Attach a [`FediverseEventObserver`] so that the fetch-and-cache /// side-effect emits granular events (SSRF rejection, actor /// impersonation) to the guard pipeline (RFC 0028). Default is the /// silent [`NoopFediverseObserver`]. #[must_use] pub fn with_observer(mut self, observer: Arc<dyn FediverseEventObserver>) -> Self { self.observer = observer; self } } impl Default for AlwaysAcceptVerifier { fn default() -> Self { Self::new() } } #[async_trait] impl SignatureVerifier for AlwaysAcceptVerifier { async fn verify(&self, ctx: &VerifyCtx<'_>) -> Result<(), AppError> { tracing::warn!( actor = %ctx.actor_ap_id, "AlwaysAcceptVerifier in use — DO NOT enable in production", ); if let Some((db, http)) = &self.resolver && let Err(e) = fetch_and_cache_remote_actor_observed( db, http, ctx.actor_ap_id, /* allow_private_ips */ true, self.observer.as_ref(), ) .await { // « Always accept » applies to signature verification, not // to the cache-side-effect: a failed fetch must not 401 the // caller. Log and let downstream services decide what to do // with an absent row. tracing::warn!( actor = %ctx.actor_ap_id, error = %e, "AlwaysAcceptVerifier: actor fetch+cache failed (continuing)", ); } Ok(()) } } /// Maximum body size accepted from a remote actor fetch. Mastodon's /// canonical Person doc is ~2 KB; 1 MiB leaves room for icons in /// `attachment` and large `summary` fields while bounding DoS exposure. pub const MAX_ACTOR_BODY_BYTES: u64 = 1024 * 1024; /// Per-request timeout applied to every outgoing actor fetch. Caps /// hangs from unreachable / slow remote instances without breaking the /// inbox handler that called the verifier. pub const ACTOR_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); /// Result of [`ssrf_guard_classified`] when it rejects: carries the IP /// the guard caught so callers can forward it to the /// [`FediverseEventObserver`]. #[derive(Debug, Clone)] pub(crate) struct SsrfRejection { pub(crate) host: String, pub(crate) attempted_ip: Option<String>, pub(crate) message: String, } impl From<SsrfRejection> for AppError { fn from(value: SsrfRejection) -> Self { AppError::Validation(value.message) } } /// Refuse to fetch URLs whose resolved IP is loopback, private, /// link-local, ULA, or unspecified. Prevents SSRF pivots through /// `actor_ap_id` values like `http://169.254.169.254/...` (AWS /// metadata) or `http://10.x.x.x/admin`. Returns a structured /// [`SsrfRejection`] on rejection so the caller can forward host + IP /// to the [`FediverseEventObserver`]. /// /// Caller passes `allow_private_ips=true` only in test/dev contexts /// where the topology is local (e.g. the docker-compose testbed using /// `*.local` aliases resolving to 172.x.x.x). async fn ssrf_guard_classified( url: &url::Url, allow_private_ips: bool, ) -> Result<(), SsrfRejection> { if allow_private_ips { return Ok(()); } let host = url.host_str().ok_or_else(|| SsrfRejection { host: String::new(), attempted_ip: None, message: "URL has no host".into(), })?; let port = url.port_or_known_default().unwrap_or(80); let addrs = tokio::net::lookup_host((host, port)) .await .map_err(|e| SsrfRejection { host: host.to_string(), attempted_ip: None, message: format!("DNS resolution failed for {host}: {e}"), })?; for addr in addrs { use std::net::IpAddr; let ip = addr.ip(); if ip.is_loopback() || ip.is_unspecified() { return Err(SsrfRejection { host: host.to_string(), attempted_ip: Some(ip.to_string()), message: format!( "refusing to fetch {host}: resolves to loopback/unspecified ({ip})" ), }); } match ip { IpAddr::V4(v4) => { if v4.is_private() || v4.is_link_local() { return Err(SsrfRejection { host: host.to_string(), attempted_ip: Some(v4.to_string()), message: format!( "refusing to fetch {host}: resolves to private/link-local ({v4})" ), }); } } IpAddr::V6(v6) => { let seg0 = v6.segments()[0]; if (seg0 & 0xfe00) == 0xfc00 || (seg0 & 0xffc0) == 0xfe80 { return Err(SsrfRejection { host: host.to_string(), attempted_ip: Some(v6.to_string()), message: format!( "refusing to fetch {host}: resolves to ULA/link-local ({v6})" ), }); } } } } Ok(()) } /// Fetch a remote AP `Person` actor over HTTP and upsert the matching /// `remote_identities` row. Idempotent: a re-fetch refreshes /// `public_key_pem`, `inbox_url`, `shared_inbox_url`, `display_name`, /// and `last_fetched_at`. /// /// Hardening: /// - 10 s per-request timeout (via the caller's `http` client). /// - 1 MiB body cap via Content-Length sniff + post-read assertion. /// - SSRF guard rejecting private/loopback/link-local IPs unless /// `allow_private_ips=true` (only set by test/dev callers). /// /// Network/parse errors return [`AppError::Validation`]; the caller /// chooses whether to surface them (production verifier) or swallow /// them (test/dev verifier). pub async fn fetch_and_cache_remote_actor( db: &DatabaseConnection, http: &reqwest::Client, actor_ap_id: &str, allow_private_ips: bool, ) -> Result<remote_identity::Model, AppError> { fetch_and_cache_remote_actor_observed( db, http, actor_ap_id, allow_private_ips, &NoopFediverseObserver, ) .await } /// Variant of [`fetch_and_cache_remote_actor`] that emits /// [`FediverseEventObserver`] events on every SSRF rejection and on /// CVE-2024-23832 actor-impersonation mismatches. pub async fn fetch_and_cache_remote_actor_observed( db: &DatabaseConnection, http: &reqwest::Client, actor_ap_id: &str, allow_private_ips: bool, observer: &dyn FediverseEventObserver, ) -> Result<remote_identity::Model, AppError> { let url = url::Url::parse(actor_ap_id) .map_err(|e| AppError::Validation(format!("actor {actor_ap_id} not a valid URL: {e}")))?; if let Err(rejection) = ssrf_guard_classified(&url, allow_private_ips).await { observer .on_fetch_ssrf(&rejection.host, rejection.attempted_ip.as_deref()) .await; return Err(AppError::from(rejection)); } let response = http .get(actor_ap_id) .header("Accept", "application/activity+json") .send() .await .map_err(|e| AppError::Validation(format!("cannot fetch actor {actor_ap_id}: {e}")))?; if !response.status().is_success() { return Err(AppError::Validation(format!( "actor fetch returned {} for {actor_ap_id}", response.status() ))); } // Belt: refuse early if the server advertises a body larger than // our cap. Suspenders: re-check after the body is fully buffered, // because Content-Length is optional and can lie. if let Some(len) = response.content_length() && len > MAX_ACTOR_BODY_BYTES { return Err(AppError::Validation(format!( "actor {actor_ap_id} body too large: {len} > {MAX_ACTOR_BODY_BYTES}" ))); } let bytes = response .bytes() .await .map_err(|e| AppError::Validation(format!("actor {actor_ap_id} read error: {e}")))?; if (bytes.len() as u64) > MAX_ACTOR_BODY_BYTES { return Err(AppError::Validation(format!( "actor {actor_ap_id} body too large: {} > {MAX_ACTOR_BODY_BYTES}", bytes.len() ))); } let body: serde_json::Value = serde_json::from_slice(&bytes) .map_err(|e| AppError::Validation(format!("actor {actor_ap_id} not valid JSON: {e}")))?; // CVE-2024-23832 parity (ANSSI F-AP-1) — refuse to cache an actor // document whose `id` does not match the URL we fetched. Without // this check, `evil.tld` could host a document claiming // `id="https://victim.tld/users/alice"` and a downstream lookup // keyed on the impersonated id would resolve to the attacker's // public key. The mismatch is also reported to the observer so // detectors can ban the lying host. if let Some(declared_id) = body.get("id").and_then(|v| v.as_str()) && declared_id != actor_ap_id { let host = url .host_str() .map(|s| s.to_ascii_lowercase()) .unwrap_or_default(); observer .on_inbox_abuse(&host, Some(actor_ap_id), "actor_impersonation") .await; return Err(AppError::Validation(format!( "actor {actor_ap_id} document `id` mismatch: claims `{declared_id}`", ))); } let preferred_username = body .get("preferredUsername") .and_then(|v| v.as_str()) .ok_or_else(|| { AppError::Validation(format!("actor {actor_ap_id} missing preferredUsername")) })? .to_string(); let domain = url .host_str() .ok_or_else(|| AppError::Validation(format!("actor {actor_ap_id} has no host")))? .to_string(); let public_key_pem = body .get("publicKey") .and_then(|v| v.get("publicKeyPem")) .and_then(|v| v.as_str()) .ok_or_else(|| { AppError::Validation(format!( "actor {actor_ap_id} missing publicKey.publicKeyPem" )) })? .to_string(); let inbox_url = body .get("inbox") .and_then(|v| v.as_str()) .ok_or_else(|| AppError::Validation(format!("actor {actor_ap_id} missing inbox")))? .to_string(); let shared_inbox_url = body .get("endpoints") .and_then(|v| v.get("sharedInbox")) .and_then(|v| v.as_str()) .map(|s| s.to_string()); let display_name = body .get("name") .and_then(|v| v.as_str()) .map(|s| s.to_string()); let now = Utc::now(); let existing = remote_identity::Entity::find() .filter(remote_identity::Column::ApId.eq(actor_ap_id)) .one(db) .await?; let model = if let Some(row) = existing { let mut active: remote_identity::ActiveModel = row.into(); active.public_key_pem = Set(public_key_pem); active.inbox_url = Set(inbox_url); active.shared_inbox_url = Set(shared_inbox_url); active.display_name = Set(display_name); active.raw_json = Set(Some(body)); active.last_fetched_at = Set(now); active.updated_at = Set(now); active.update(db).await? } else { let active = remote_identity::ActiveModel { id: Set(Uuid::new_v4()), ap_id: Set(actor_ap_id.to_string()), username: Set(preferred_username), domain: Set(domain), display_name: Set(display_name), summary_html: Set(None), inbox_url: Set(inbox_url), shared_inbox_url: Set(shared_inbox_url), public_key_pem: Set(public_key_pem), icon_url: Set(None), actor_type: Set("Person".to_string()), raw_json: Set(Some(body)), last_fetched_at: Set(now), disabled_at: Set(None), created_at: Set(now), updated_at: Set(now), }; active.insert(db).await? }; Ok(model) } fn parse_handle_from_ap_id(ap_id: &str) -> Option<(String, String)> { let url = url::Url::parse(ap_id).ok()?; let host = url.host_str()?.to_string(); let user = url.path_segments()?.next_back()?.to_string(); Some((user, host)) } #[cfg(test)] mod tests { use super::*; use sea_orm::{DatabaseBackend, MockDatabase, MockExecResult}; fn fresh_remote(id: Uuid, ap_id: &str) -> remote_identity::Model { let now = Utc::now(); remote_identity::Model { id, ap_id: ap_id.to_string(), username: "alice".into(), domain: "a.tld".into(), display_name: Some("Alice".into()), summary_html: None, inbox_url: format!("{ap_id}/inbox"), shared_inbox_url: None, public_key_pem: "-----BEGIN PUBLIC KEY-----\n".into(), icon_url: None, actor_type: "Person".into(), raw_json: None, last_fetched_at: now, disabled_at: None, created_at: now, updated_at: now, } } // ─── Pure ──────────────────────────────────────────────────────── #[test] fn delete_outcome_default_is_zero() { assert_eq!(DeleteOutcome::default().rows_disabled, 0); assert_eq!(DeleteOutcome::default().jwts_blacklisted, 0); } // ─── apply_update_actor ───────────────────────────────────────── #[tokio::test] async fn apply_update_actor_upserts_when_existing() { let id = Uuid::new_v4(); let row = fresh_remote(id, "https://a.tld/users/alice"); let mut after = row.clone(); after.display_name = Some("Alice McUpdated".into()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .append_query_results([vec![after]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let updated = FediverseSignalService::apply_update_actor( &db, UpdatedActor { ap_id: "https://a.tld/users/alice".into(), display_name: Some("Alice McUpdated".into()), summary_html: None, icon_url: None, public_key_pem: None, raw_json: None, }, ) .await .unwrap(); assert_eq!(updated.display_name.as_deref(), Some("Alice McUpdated")); } #[tokio::test] async fn apply_update_actor_inserts_when_missing() { let id = Uuid::new_v4(); let new_row = fresh_remote(id, "https://a.tld/users/bob"); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) // 1. SELECT existing → empty .append_query_results([Vec::<remote_identity::Model>::new()]) // 2. INSERT returns the row .append_query_results([vec![new_row.clone()]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let inserted = FediverseSignalService::apply_update_actor( &db, UpdatedActor { ap_id: "https://a.tld/users/bob".into(), display_name: None, summary_html: None, icon_url: None, public_key_pem: None, raw_json: None, }, ) .await .unwrap(); assert_eq!(inserted.ap_id, new_row.ap_id); } // ─── apply_delete_actor ───────────────────────────────────────── #[tokio::test] async fn apply_delete_actor_sets_disabled_at_and_returns_one() { let id = Uuid::new_v4(); let row = fresh_remote(id, "https://a.tld/users/alice"); let mut after = row.clone(); after.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .append_query_results([vec![after]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let outcome = FediverseSignalService::apply_delete_actor(&db, "https://a.tld/users/alice") .await .unwrap(); assert_eq!(outcome.rows_disabled, 1); assert_eq!(outcome.jwts_blacklisted, 0); } #[tokio::test] async fn apply_delete_actor_returns_zero_when_no_row() { let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([Vec::<remote_identity::Model>::new()]) .into_connection(); let outcome = FediverseSignalService::apply_delete_actor(&db, "https://a.tld/users/ghost") .await .unwrap(); assert_eq!(outcome, DeleteOutcome::default()); } #[tokio::test] async fn apply_delete_actor_returns_zero_when_already_disabled() { let id = Uuid::new_v4(); let mut row = fresh_remote(id, "https://a.tld/users/alice"); row.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .into_connection(); let outcome = FediverseSignalService::apply_delete_actor(&db, "https://a.tld/users/alice") .await .unwrap(); assert_eq!(outcome.rows_disabled, 0); } // ─── apply_delete_actor_observed (RFC 0032) ───────────────────── #[derive(Default)] struct CapturingDeleteObserver { calls: std::sync::Mutex<Vec<(String, String, Uuid)>>, } #[async_trait] impl FediverseEventObserver for CapturingDeleteObserver { async fn on_delete_actor(&self, domain: &str, ap_id: &str, remote_identity_id: Uuid) { self.calls.lock().unwrap().push(( domain.to_string(), ap_id.to_string(), remote_identity_id, )); } } #[tokio::test] async fn apply_delete_actor_observed_fires_hook_on_first_disable() { let id = Uuid::new_v4(); let row = fresh_remote(id, "https://a.tld/users/alice"); let mut after = row.clone(); after.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .append_query_results([vec![after]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let observer = CapturingDeleteObserver::default(); let outcome = FediverseSignalService::apply_delete_actor_observed( &db, "https://a.tld/users/alice", &observer, ) .await .unwrap(); assert_eq!(outcome.rows_disabled, 1); let calls = observer.calls.lock().unwrap().clone(); assert_eq!(calls.len(), 1); assert_eq!(calls[0].0, "a.tld"); assert_eq!(calls[0].1, "https://a.tld/users/alice"); assert_eq!(calls[0].2, id); } #[tokio::test] async fn apply_delete_actor_observed_does_not_fire_when_no_row() { let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([Vec::<remote_identity::Model>::new()]) .into_connection(); let observer = CapturingDeleteObserver::default(); let outcome = FediverseSignalService::apply_delete_actor_observed( &db, "https://a.tld/users/ghost", &observer, ) .await .unwrap(); assert_eq!(outcome, DeleteOutcome::default()); assert!(observer.calls.lock().unwrap().is_empty()); } #[tokio::test] async fn apply_delete_actor_observed_does_not_fire_when_already_disabled() { let id = Uuid::new_v4(); let mut row = fresh_remote(id, "https://a.tld/users/alice"); row.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .into_connection(); let observer = CapturingDeleteObserver::default(); let outcome = FediverseSignalService::apply_delete_actor_observed( &db, "https://a.tld/users/alice", &observer, ) .await .unwrap(); assert_eq!(outcome.rows_disabled, 0); assert!(observer.calls.lock().unwrap().is_empty()); } // ─── apply_revoke_membership ──────────────────────────────────── #[tokio::test] async fn apply_revoke_membership_returns_false_when_no_row() { let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 0, }]) .into_connection(); let revoked = FediverseSignalService::apply_revoke_membership(&db, Uuid::new_v4(), Uuid::new_v4()) .await .unwrap(); assert!(!revoked); } #[tokio::test] async fn apply_revoke_membership_returns_true_when_row_dropped() { let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let revoked = FediverseSignalService::apply_revoke_membership(&db, Uuid::new_v4(), Uuid::new_v4()) .await .unwrap(); assert!(revoked); } // ─── SignatureVerifier ────────────────────────────────────────── fn dummy_ctx<'a>(headers: &'a axum::http::HeaderMap) -> VerifyCtx<'a> { VerifyCtx { actor_ap_id: "https://a.tld/users/alice", method: "post", path: "/inbox", headers, body: b"", public_key_pem: "", } } #[tokio::test] async fn always_accept_verifier_accepts_any_input() { let headers = axum::http::HeaderMap::new(); AlwaysAcceptVerifier::new() .verify(&dummy_ctx(&headers)) .await .expect("AlwaysAcceptVerifier never rejects"); } #[tokio::test] async fn signature_verifier_trait_object_works() { let verifier: Box<dyn SignatureVerifier> = Box::new(AlwaysAcceptVerifier::new()); let headers = axum::http::HeaderMap::new(); verifier .verify(&dummy_ctx(&headers)) .await .expect("trait object dispatches"); } // ─── RFC 0018 — apply_delete_actor_with_iterator ──────────────── #[tokio::test] async fn apply_delete_actor_with_noop_iterator_keeps_jwts_blacklisted_zero() { use rustwarden_core::services::jwt_blacklist_service::NoopActiveTokenIterator; let id = Uuid::new_v4(); let row = fresh_remote(id, "https://a.tld/users/alice"); let mut after = row.clone(); after.disabled_at = Some(Utc::now()); // RFC 0033: the unified path needs only SELECT + UPDATE (no // re-fetch); the iterator is a Noop so no blacklist write. let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .append_query_results([vec![after]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let outcome = FediverseSignalService::apply_delete_actor_with_iterator( &db, "https://a.tld/users/alice", &NoopActiveTokenIterator, ) .await .unwrap(); assert_eq!(outcome.rows_disabled, 1); assert_eq!(outcome.jwts_blacklisted, 0); } // ─── apply_delete_actor_with_hooks (RFC 0033) ─────────────────── #[test] fn delete_actor_hooks_default_is_all_none() { let hooks = DeleteActorHooks::default(); assert!(hooks.observer.is_none()); assert!(hooks.iterator.is_none()); } #[tokio::test] async fn apply_delete_actor_with_hooks_combines_observer_and_iterator() { use rustwarden_core::services::jwt_blacklist_service::NoopActiveTokenIterator; let id = Uuid::new_v4(); let row = fresh_remote(id, "https://a.tld/users/alice"); let mut after = row.clone(); after.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .append_query_results([vec![after]]) .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let observer = CapturingDeleteObserver::default(); let iterator = NoopActiveTokenIterator; let hooks = DeleteActorHooks { observer: Some(&observer), iterator: Some(&iterator), }; let outcome = FediverseSignalService::apply_delete_actor_with_hooks( &db, "https://a.tld/users/alice", hooks, ) .await .unwrap(); assert_eq!(outcome.rows_disabled, 1); assert_eq!(outcome.jwts_blacklisted, 0); let calls = observer.calls.lock().unwrap().clone(); assert_eq!(calls.len(), 1); assert_eq!(calls[0].2, id); } #[tokio::test] async fn apply_delete_actor_with_hooks_skips_both_when_already_disabled() { use rustwarden_core::services::jwt_blacklist_service::NoopActiveTokenIterator; let id = Uuid::new_v4(); let mut row = fresh_remote(id, "https://a.tld/users/alice"); row.disabled_at = Some(Utc::now()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![row]]) .into_connection(); let observer = CapturingDeleteObserver::default(); let iterator = NoopActiveTokenIterator; let outcome = FediverseSignalService::apply_delete_actor_with_hooks( &db, "https://a.tld/users/alice", DeleteActorHooks { observer: Some(&observer), iterator: Some(&iterator), }, ) .await .unwrap(); assert_eq!(outcome.rows_disabled, 0); assert_eq!(outcome.jwts_blacklisted, 0); assert!(observer.calls.lock().unwrap().is_empty()); } #[tokio::test] async fn noop_active_token_iterator_returns_empty_list() { use rustwarden_core::services::jwt_blacklist_service::{ ActiveTokenIterator, NoopActiveTokenIterator, }; let v = NoopActiveTokenIterator .active_tokens_for_remote(Uuid::new_v4()) .await .unwrap(); assert!(v.is_empty()); } // ─── fetch_and_cache_remote_actor (F2 hardening) ──────────────── /// Build a syntactically valid AP Person doc with the given overrides. /// `extra` is merged at the top level — pass `null` for « no extra ». fn ap_person_json( actor_url: &str, username: &str, with_public_key: bool, with_inbox: bool, ) -> serde_json::Value { let mut doc = serde_json::json!({ "@context": ["https://www.w3.org/ns/activitystreams"], "type": "Person", "id": actor_url, "preferredUsername": username, "name": username, }); if with_public_key { doc["publicKey"] = serde_json::json!({ "id": format!("{actor_url}#main-key"), "owner": actor_url, "publicKeyPem": "-----BEGIN PUBLIC KEY-----\nFAKE\n-----END PUBLIC KEY-----\n", }); } if with_inbox { doc["inbox"] = serde_json::json!(format!("{actor_url}/inbox")); } doc } fn testbed_http_client() -> reqwest::Client { reqwest::Client::builder() .timeout(std::time::Duration::from_secs(2)) .build() .unwrap() } #[tokio::test] async fn fetch_inserts_new_remote_identity_on_first_sight() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/alice", server.uri()); Mock::given(method("GET")) .and(path("/users/alice")) .respond_with( ResponseTemplate::new(200) .set_body_json(ap_person_json(&actor_url, "alice", true, true)), ) .mount(&server) .await; let inserted = fresh_remote(Uuid::new_v4(), &actor_url); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([Vec::<remote_identity::Model>::new()]) // SELECT empty .append_query_results([vec![inserted.clone()]]) // INSERT RETURNING .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); let model = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect("happy path INSERT must succeed"); assert_eq!(model.ap_id, inserted.ap_id); } #[tokio::test] async fn fetch_updates_existing_remote_identity_on_refetch() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/bob", server.uri()); Mock::given(method("GET")) .and(path("/users/bob")) .respond_with( ResponseTemplate::new(200) .set_body_json(ap_person_json(&actor_url, "bob", true, true)), ) .mount(&server) .await; let existing = fresh_remote(Uuid::new_v4(), &actor_url); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres) .append_query_results([vec![existing.clone()]]) // SELECT hit .append_query_results([vec![existing.clone()]]) // UPDATE RETURNING .append_exec_results([MockExecResult { last_insert_id: 0, rows_affected: 1, }]) .into_connection(); fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect("happy path UPDATE must succeed"); } #[tokio::test] async fn fetch_returns_error_on_http_404() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/ghost", server.uri()); Mock::given(method("GET")) .and(path("/users/ghost")) .respond_with(ResponseTemplate::new(404)) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect_err("404 must be surfaced as Validation"); assert!(format!("{err}").contains("404"), "got: {err}"); } #[tokio::test] async fn fetch_returns_error_on_malformed_json() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/junk", server.uri()); Mock::given(method("GET")) .and(path("/users/junk")) .respond_with(ResponseTemplate::new(200).set_body_string("not-json-at-all")) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect_err("malformed JSON must be rejected"); assert!(format!("{err}").contains("not valid JSON"), "got: {err}"); } #[tokio::test] async fn fetch_returns_error_when_public_key_missing() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/nokey", server.uri()); Mock::given(method("GET")) .and(path("/users/nokey")) .respond_with( ResponseTemplate::new(200) .set_body_json(ap_person_json(&actor_url, "nokey", false, true)), ) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect_err("missing publicKey must be rejected"); assert!(format!("{err}").contains("publicKey"), "got: {err}"); } #[tokio::test] async fn fetch_returns_error_when_inbox_missing() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/noinbox", server.uri()); Mock::given(method("GET")) .and(path("/users/noinbox")) .respond_with( ResponseTemplate::new(200) .set_body_json(ap_person_json(&actor_url, "noinbox", true, false)), ) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect_err("missing inbox must be rejected"); assert!(format!("{err}").contains("inbox"), "got: {err}"); } #[tokio::test] async fn fetch_returns_error_on_invalid_url() { let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), "not a valid url", true) .await .expect_err("malformed URL must be rejected before HTTP"); assert!(format!("{err}").contains("not a valid URL"), "got: {err}"); } #[tokio::test] async fn fetch_returns_error_when_body_exceeds_cap() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/huge", server.uri()); // Send a payload just over our cap. wiremock sets a truthful // Content-Length, so the pre-flight check trips before we // allocate the full body in our process. let payload = vec![b'x'; (MAX_ACTOR_BODY_BYTES as usize) + 1]; Mock::given(method("GET")) .and(path("/users/huge")) .respond_with(ResponseTemplate::new(200).set_body_bytes(payload)) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor(&db, &testbed_http_client(), &actor_url, true) .await .expect_err("oversized body must be rejected"); assert!(format!("{err}").contains("too large"), "got: {err}"); } #[tokio::test] async fn ssrf_guard_rejects_loopback_when_strict() { // Use a known-loopback URL (wiremock binds 127.0.0.1, so the // server starts but the guard must reject before HTTP fires). use wiremock::MockServer; let server = MockServer::start().await; let actor_url = format!("{}/users/anything", server.uri()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let err = fetch_and_cache_remote_actor( &db, &testbed_http_client(), &actor_url, /* allow_private_ips */ false, ) .await .expect_err("loopback must be refused when allow_private_ips=false"); let msg = format!("{err}"); assert!( msg.contains("loopback") || msg.contains("private") || msg.contains("link-local"), "expected SSRF guard rejection, got: {err}" ); } // ─── RFC 0028 / Lot 7b — observer emission points ────────────── use std::sync::atomic::{AtomicUsize, Ordering}; #[derive(Default)] struct CountingObserver { ssrf: AtomicUsize, inbox_abuse_reason: std::sync::Mutex<Vec<String>>, } #[async_trait] impl FediverseEventObserver for CountingObserver { async fn on_fetch_ssrf(&self, _host: &str, _ip: Option<&str>) { self.ssrf.fetch_add(1, Ordering::SeqCst); } async fn on_inbox_abuse(&self, _domain: &str, _actor: Option<&str>, reason: &str) { self.inbox_abuse_reason .lock() .unwrap() .push(reason.to_string()); } } #[tokio::test] async fn fetch_emits_ssrf_event_on_loopback_rejection() { use wiremock::MockServer; let server = MockServer::start().await; let actor_url = format!("{}/users/anything", server.uri()); let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let observer = CountingObserver::default(); let err = fetch_and_cache_remote_actor_observed( &db, &testbed_http_client(), &actor_url, /* allow_private_ips */ false, &observer, ) .await .expect_err("loopback must reject"); assert!(matches!(err, AppError::Validation(_))); assert_eq!(observer.ssrf.load(Ordering::SeqCst), 1); } #[tokio::test] async fn fetch_emits_actor_impersonation_event_on_id_mismatch() { use wiremock::matchers::{method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; let server = MockServer::start().await; let actor_url = format!("{}/users/alice", server.uri()); // The document claims a different `id` — CVE-2024-23832 parity. let mut doc = ap_person_json(&actor_url, "alice", true, true); doc["id"] = serde_json::json!("https://victim.tld/users/alice"); Mock::given(method("GET")) .and(path("/users/alice")) .respond_with(ResponseTemplate::new(200).set_body_json(doc)) .mount(&server) .await; let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres).into_connection(); let observer = CountingObserver::default(); let err = fetch_and_cache_remote_actor_observed( &db, &testbed_http_client(), &actor_url, true, &observer, ) .await .expect_err("id mismatch must reject"); assert!(format!("{err}").contains("mismatch"), "got {err}"); let reasons = observer.inbox_abuse_reason.lock().unwrap().clone(); assert_eq!(reasons, vec!["actor_impersonation".to_string()]); } }