GitRust GitRust
Connexion

eric / rustwarden

start fediverse guard

EBO <eric.bouhana@softalys.com> committé le 2026-05-14 21:29

2902d73364688b47b2681b9494ec2d041eb5d0bb

1 parent(s)

13 fichiers modifiés +1579 -8
M Cargo.lock
+47 -8
@@ -1245,6 +1245,20 @@ dependencies = [
1245 1245 ]
1246 1246
1247 1247 [[package]]
1248 +name = "dashmap"
1249 +version = "6.1.0"
1250 +source = "registry+https://github.com/rust-lang/crates.io-index"
1251 +checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf"
1252 +dependencies = [
1253 + "cfg-if",
1254 + "crossbeam-utils",
1255 + "hashbrown 0.14.5",
1256 + "lock_api",
1257 + "once_cell",
1258 + "parking_lot_core",
1259 +]
1260 +
1261 +[[package]]
1248 1262 name = "deadpool"
1249 1263 version = "0.12.3"
1250 1264 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1536,7 +1550,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
1536 1550 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
1537 1551 dependencies = [
1538 1552 "libc",
1539 - "windows-sys 0.52.0",
1553 + "windows-sys 0.61.2",
1540 1554 ]
1541 1555
1542 1556 [[package]]
@@ -1962,6 +1976,12 @@ dependencies = [
1962 1976
1963 1977 [[package]]
1964 1978 name = "hashbrown"
1979 +version = "0.14.5"
1980 +source = "registry+https://github.com/rust-lang/crates.io-index"
1981 +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
1982 +
1983 +[[package]]
1984 +name = "hashbrown"
1965 1985 version = "0.15.5"
1966 1986 source = "registry+https://github.com/rust-lang/crates.io-index"
1967 1987 checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
@@ -2220,7 +2240,7 @@ dependencies = [
2220 2240 "libc",
2221 2241 "percent-encoding",
2222 2242 "pin-project-lite",
2223 - "socket2 0.5.10",
2243 + "socket2 0.6.3",
2224 2244 "system-configuration",
2225 2245 "tokio",
2226 2246 "tower-layer",
@@ -3046,7 +3066,7 @@ version = "5.0.0"
3046 3066 source = "registry+https://github.com/rust-lang/crates.io-index"
3047 3067 checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
3048 3068 dependencies = [
3049 - "base64 0.21.7",
3069 + "base64 0.22.1",
3050 3070 "chrono",
3051 3071 "getrandom 0.2.17",
3052 3072 "http",
@@ -3578,7 +3598,7 @@ dependencies = [
3578 3598 "quinn-udp",
3579 3599 "rustc-hash",
3580 3600 "rustls",
3581 - "socket2 0.5.10",
3601 + "socket2 0.6.3",
3582 3602 "thiserror 2.0.18",
3583 3603 "tokio",
3584 3604 "tracing",
@@ -3615,9 +3635,9 @@ dependencies = [
3615 3635 "cfg_aliases",
3616 3636 "libc",
3617 3637 "once_cell",
3618 - "socket2 0.5.10",
3638 + "socket2 0.6.3",
3619 3639 "tracing",
3620 - "windows-sys 0.52.0",
3640 + "windows-sys 0.59.0",
3621 3641 ]
3622 3642
3623 3643 [[package]]
@@ -4027,7 +4047,7 @@ dependencies = [
4027 4047 "errno",
4028 4048 "libc",
4029 4049 "linux-raw-sys",
4030 - "windows-sys 0.52.0",
4050 + "windows-sys 0.61.2",
4031 4051 ]
4032 4052
4033 4053 [[package]]
@@ -4189,6 +4209,25 @@ dependencies = [
4189 4209 ]
4190 4210
4191 4211 [[package]]
4212 +name = "rustwarden-fediverse-guard"
4213 +version = "0.1.0"
4214 +dependencies = [
4215 + "async-trait",
4216 + "chrono",
4217 + "dashmap",
4218 + "rustwarden-core",
4219 + "rustwarden-fediverse",
4220 + "sea-orm",
4221 + "sea-orm-migration",
4222 + "serde",
4223 + "serde_json",
4224 + "thiserror 2.0.18",
4225 + "tokio",
4226 + "tracing",
4227 + "uuid",
4228 +]
4229 +
4230 +[[package]]
4192 4231 name = "rustwarden-fediverse-ui"
4193 4232 version = "0.1.0"
4194 4233 dependencies = [
@@ -5125,7 +5164,7 @@ dependencies = [
5125 5164 "getrandom 0.4.2",
5126 5165 "once_cell",
5127 5166 "rustix",
5128 - "windows-sys 0.52.0",
5167 + "windows-sys 0.61.2",
5129 5168 ]
5130 5169
5131 5170 [[package]]
M Cargo.toml
+1 -0
@@ -2,6 +2,7 @@
2 2 members = [
3 3 "crates/rustwarden-core",
4 4 "crates/rustwarden-fediverse",
5 + "crates/rustwarden-fediverse-guard",
5 6 "crates/rustwarden-fediverse-ui",
6 7 "crates/rustwarden-ui",
7 8 "app",
A crates/rustwarden-fediverse-guard/Cargo.toml
+44 -0
@@ -0,0 +1,44 @@
1 +[package]
2 +name = "rustwarden-fediverse-guard"
3 +version = "0.1.0"
4 +edition.workspace = true
5 +rust-version.workspace = true
6 +description = "Real-time abuse detection and ban management for the Rustwarden ActivityPub federation layer (RFC 0028+)"
7 +
8 +[features]
9 +default = ["ui"]
10 +# Reserved for the admin UI module (handlers + templates).
11 +ui = []
12 +
13 +[dependencies]
14 +rustwarden-core = { path = "../rustwarden-core" }
15 +rustwarden-fediverse = { path = "../rustwarden-fediverse" }
16 +# Async runtime
17 +tokio = { version = "1", features = ["sync", "time", "rt", "macros"] }
18 +async-trait = "0.1"
19 +# ORM + migrations
20 +sea-orm = { version = "2.0.0-rc.20", features = ["sqlx-postgres", "runtime-tokio-native-tls", "macros"] }
21 +sea-orm-migration = { version = "2.0.0-rc.20", features = ["sqlx-postgres", "runtime-tokio-native-tls"] }
22 +# Concurrent hot-path
23 +dashmap = "6"
24 +# Serialization
25 +serde = { version = "1.0", features = ["derive"] }
26 +serde_json = "1.0"
27 +# Identifiers / time
28 +uuid = { version = "1.0", features = ["v4", "serde"] }
29 +chrono = { version = "0.4", features = ["serde"] }
30 +# Errors
31 +thiserror = "2.0"
32 +# Telemetry
33 +tracing = "0.1"
34 +
35 +[lints]
36 +workspace = true
37 +
38 +[dev-dependencies]
39 +sea-orm = { version = "2.0.0-rc.20", features = ["mock"] }
40 +tokio = { version = "1", features = ["full", "test-util"] }
41 +
42 +[[bin]]
43 +name = "migrate-fediverse-guard"
44 +path = "migration/src/main.rs"
A crates/rustwarden-fediverse-guard/migration/src/lib.rs
+39 -0
@@ -0,0 +1,39 @@
1 +//! SeaORM migrations for the rustwarden-fediverse-guard crate.
2 +//!
3 +//! Single fused migration that creates the three tables backing the
4 +//! guard runtime:
5 +//!
6 +//! - `fediverse_guard_events` (raw event log, sliding-window source)
7 +//! - `fediverse_guard_bans` (active + historical bans)
8 +//! - `fediverse_guard_acl` (manual allow / deny rules)
9 +//!
10 +//! The migrator uses a **dedicated** tracking table
11 +//! `seaql_fediverse_guard_migrations` so it can be applied
12 +//! independently of the core migrator.
13 +//!
14 +//! Run after both core and fediverse migrators:
15 +//! ```bash
16 +//! cargo run --bin migrate -- up
17 +//! cargo run -p rustwarden-fediverse --bin migrate-fediverse -- up
18 +//! cargo run -p rustwarden-fediverse-guard --bin migrate-fediverse-guard -- up
19 +//! ```
20 +
21 +pub use sea_orm_migration::prelude::*;
22 +
23 +pub mod m20260514_000001_fediverse_guard_schema;
24 +
25 +/// Migrator for the rustwarden-fediverse-guard crate.
26 +pub struct FediverseGuardMigrator;
27 +
28 +#[async_trait::async_trait]
29 +impl MigratorTrait for FediverseGuardMigrator {
30 + fn migration_table_name() -> sea_orm::DynIden {
31 + Alias::new("seaql_fediverse_guard_migrations").into_iden()
32 + }
33 +
34 + fn migrations() -> Vec<Box<dyn MigrationTrait>> {
35 + vec![Box::new(
36 + m20260514_000001_fediverse_guard_schema::Migration,
37 + )]
38 + }
39 +}
A crates/rustwarden-fediverse-guard/migration/src/m20260514_000001_fediverse_guard_schema.rs
+262 -0
@@ -0,0 +1,262 @@
1 +//! Migration `m20260514_000001_fediverse_guard_schema`.
2 +//!
3 +//! Creates the three tables backing the fediverse-guard runtime:
4 +//!
5 +//! - `fediverse_guard_events` — raw event log, indexed by
6 +//! `(domain, created_at DESC)` to feed sliding-window detectors.
7 +//! - `fediverse_guard_bans` — active + historical bans, indexed by
8 +//! `(domain)` with a partial-unique on rows whose `lifted_at IS NULL`.
9 +//! - `fediverse_guard_acl` — manual allow/deny rules with FK to
10 +//! `users.id`.
11 +//!
12 +//! All columns use UTC timestamps. UUID primary keys are caller-
13 +//! generated (Rust-side `uuid::Uuid::new_v4()`) — Postgres' `gen_random_uuid()`
14 +//! is reserved for the rare DDL-only inserts (none here).
15 +
16 +use sea_orm_migration::prelude::*;
17 +use sea_orm_migration::sea_orm::ConnectionTrait;
18 +
19 +#[derive(DeriveMigrationName)]
20 +pub struct Migration;
21 +
22 +#[async_trait::async_trait]
23 +impl MigrationTrait for Migration {
24 + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
25 + // ---- fediverse_guard_events ----------------------------------
26 + manager
27 + .create_table(
28 + Table::create()
29 + .table(FediverseGuardEvents::Table)
30 + .col(
31 + ColumnDef::new(FediverseGuardEvents::Id)
32 + .uuid()
33 + .not_null()
34 + .primary_key(),
35 + )
36 + .col(
37 + ColumnDef::new(FediverseGuardEvents::EventType)
38 + .text()
39 + .not_null(),
40 + )
41 + .col(
42 + ColumnDef::new(FediverseGuardEvents::Domain)
43 + .text()
44 + .not_null(),
45 + )
46 + .col(ColumnDef::new(FediverseGuardEvents::ActorUri).text().null())
47 + .col(
48 + ColumnDef::new(FediverseGuardEvents::RemoteIdentityId)
49 + .uuid()
50 + .null(),
51 + )
52 + .col(
53 + ColumnDef::new(FediverseGuardEvents::Severity)
54 + .text()
55 + .not_null(),
56 + )
57 + .col(
58 + ColumnDef::new(FediverseGuardEvents::HttpStatus)
59 + .integer()
60 + .null(),
61 + )
62 + .col(
63 + ColumnDef::new(FediverseGuardEvents::Details)
64 + .json_binary()
65 + .null(),
66 + )
67 + .col(
68 + ColumnDef::new(FediverseGuardEvents::CreatedAt)
69 + .timestamp_with_time_zone()
70 + .not_null(),
71 + )
72 + .to_owned(),
73 + )
74 + .await?;
75 +
76 + let conn = manager.get_connection();
77 + conn.execute_unprepared(
78 + "CREATE INDEX fediverse_guard_events_domain_time_idx \
79 + ON fediverse_guard_events (domain, created_at DESC)",
80 + )
81 + .await?;
82 + conn.execute_unprepared(
83 + "CREATE INDEX fediverse_guard_events_type_time_idx \
84 + ON fediverse_guard_events (event_type, created_at DESC)",
85 + )
86 + .await?;
87 +
88 + // ---- fediverse_guard_bans ------------------------------------
89 + manager
90 + .create_table(
91 + Table::create()
92 + .table(FediverseGuardBans::Table)
93 + .col(
94 + ColumnDef::new(FediverseGuardBans::Id)
95 + .uuid()
96 + .not_null()
97 + .primary_key(),
98 + )
99 + .col(
100 + ColumnDef::new(FediverseGuardBans::Domain)
101 + .text()
102 + .not_null(),
103 + )
104 + .col(ColumnDef::new(FediverseGuardBans::ActorUri).text().null())
105 + .col(
106 + ColumnDef::new(FediverseGuardBans::Reason)
107 + .text()
108 + .not_null(),
109 + )
110 + .col(
111 + ColumnDef::new(FediverseGuardBans::Severity)
112 + .text()
113 + .not_null(),
114 + )
115 + .col(
116 + ColumnDef::new(FediverseGuardBans::TriggeredBy)
117 + .text()
118 + .not_null(),
119 + )
120 + .col(
121 + ColumnDef::new(FediverseGuardBans::BannedAt)
122 + .timestamp_with_time_zone()
123 + .not_null(),
124 + )
125 + .col(
126 + ColumnDef::new(FediverseGuardBans::ExpiresAt)
127 + .timestamp_with_time_zone()
128 + .null(),
129 + )
130 + .col(
131 + ColumnDef::new(FediverseGuardBans::LiftedAt)
132 + .timestamp_with_time_zone()
133 + .null(),
134 + )
135 + .col(
136 + ColumnDef::new(FediverseGuardBans::LiftedBy)
137 + .uuid()
138 + .null(),
139 + )
140 + .to_owned(),
141 + )
142 + .await?;
143 +
144 + conn.execute_unprepared(
145 + "CREATE INDEX fediverse_guard_bans_active_idx \
146 + ON fediverse_guard_bans (domain) WHERE lifted_at IS NULL",
147 + )
148 + .await?;
149 +
150 + // ---- fediverse_guard_acl -------------------------------------
151 + manager
152 + .create_table(
153 + Table::create()
154 + .table(FediverseGuardAcl::Table)
155 + .col(
156 + ColumnDef::new(FediverseGuardAcl::Id)
157 + .uuid()
158 + .not_null()
159 + .primary_key(),
160 + )
161 + .col(
162 + ColumnDef::new(FediverseGuardAcl::Kind)
163 + .text()
164 + .not_null(),
165 + )
166 + .col(
167 + ColumnDef::new(FediverseGuardAcl::Domain)
168 + .text()
169 + .not_null(),
170 + )
171 + .col(ColumnDef::new(FediverseGuardAcl::Reason).text().null())
172 + .col(
173 + ColumnDef::new(FediverseGuardAcl::CreatedBy)
174 + .uuid()
175 + .not_null(),
176 + )
177 + .col(
178 + ColumnDef::new(FediverseGuardAcl::CreatedAt)
179 + .timestamp_with_time_zone()
180 + .not_null(),
181 + )
182 + .col(
183 + ColumnDef::new(FediverseGuardAcl::UpdatedAt)
184 + .timestamp_with_time_zone()
185 + .not_null(),
186 + )
187 + .to_owned(),
188 + )
189 + .await?;
190 +
191 + conn.execute_unprepared(
192 + "CREATE UNIQUE INDEX fediverse_guard_acl_kind_domain_uniq \
193 + ON fediverse_guard_acl (kind, domain)",
194 + )
195 + .await?;
196 +
197 + Ok(())
198 + }
199 +
200 + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
201 + let conn = manager.get_connection();
202 + conn.execute_unprepared("DROP INDEX IF EXISTS fediverse_guard_acl_kind_domain_uniq")
203 + .await?;
204 + manager
205 + .drop_table(Table::drop().table(FediverseGuardAcl::Table).to_owned())
206 + .await?;
207 + conn.execute_unprepared("DROP INDEX IF EXISTS fediverse_guard_bans_active_idx")
208 + .await?;
209 + manager
210 + .drop_table(Table::drop().table(FediverseGuardBans::Table).to_owned())
211 + .await?;
212 + conn.execute_unprepared("DROP INDEX IF EXISTS fediverse_guard_events_type_time_idx")
213 + .await?;
214 + conn.execute_unprepared("DROP INDEX IF EXISTS fediverse_guard_events_domain_time_idx")
215 + .await?;
216 + manager
217 + .drop_table(Table::drop().table(FediverseGuardEvents::Table).to_owned())
218 + .await?;
219 + Ok(())
220 + }
221 +}
222 +
223 +#[derive(DeriveIden)]
224 +enum FediverseGuardEvents {
225 + Table,
226 + Id,
227 + EventType,
228 + Domain,
229 + ActorUri,
230 + RemoteIdentityId,
231 + Severity,
232 + HttpStatus,
233 + Details,
234 + CreatedAt,
235 +}
236 +
237 +#[derive(DeriveIden)]
238 +enum FediverseGuardBans {
239 + Table,
240 + Id,
241 + Domain,
242 + ActorUri,
243 + Reason,
244 + Severity,
245 + TriggeredBy,
246 + BannedAt,
247 + ExpiresAt,
248 + LiftedAt,
249 + LiftedBy,
250 +}
251 +
252 +#[derive(DeriveIden)]
253 +enum FediverseGuardAcl {
254 + Table,
255 + Id,
256 + Kind,
257 + Domain,
258 + Reason,
259 + CreatedBy,
260 + CreatedAt,
261 + UpdatedAt,
262 +}
A crates/rustwarden-fediverse-guard/migration/src/main.rs
+11 -0
@@ -0,0 +1,11 @@
1 +//! CLI entry point for SeaORM migrations of the rustwarden-fediverse-guard crate.
2 +
3 +use sea_orm_migration::prelude::*;
4 +
5 +#[path = "lib.rs"]
6 +mod migration;
7 +
8 +#[tokio::main]
9 +async fn main() {
10 + cli::run_cli(migration::FediverseGuardMigrator).await;
11 +}
A crates/rustwarden-fediverse-guard/src/ban.rs
+312 -0
@@ -0,0 +1,312 @@
1 +//! Ban-decision engine on top of the [`GuardStore`].
2 +//!
3 +//! Composition is:
4 +//!
5 +//! 1. ACL `Deny` rule → ban
6 +//! 2. Active `BanEntry` in store → ban
7 +//! 3. ACL `Allow` rule → never ban (defence in depth even if a
8 +//! detector forgets to check)
9 +//! 4. otherwise → allow
10 +//!
11 +//! Auto-ban requests pass through [`BanManager::auto_ban`], which
12 +//! honours the dry-run flag of [`crate::GuardConfig`] and refuses to
13 +//! ban allow-listed domains.
14 +
15 +use std::sync::Arc;
16 +
17 +use chrono::Utc;
18 +use rustwarden_core::error::AppError;
19 +use uuid::Uuid;
20 +
21 +use crate::config::GuardConfig;
22 +use crate::events::{GuardEvent, GuardEventKind, Severity};
23 +use crate::store::{AclEntry, AclKind, BanEntry, GuardStore};
24 +
25 +/// Effective decision for one (domain, actor_uri?) pair.
26 +#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27 +pub enum BanDecision {
28 + /// Allow the request (no ban, no deny rule).
29 + Allow,
30 + /// Deny the request (active ban or deny ACL).
31 + Deny,
32 +}
33 +
34 +/// Wraps a [`GuardStore`] with the [`GuardConfig`] runtime knobs and
35 +/// exposes a small surface used by handlers and detectors.
36 +#[derive(Clone)]
37 +pub struct BanManager {
38 + store: Arc<dyn GuardStore>,
39 + config: GuardConfig,
40 +}
41 +
42 +impl BanManager {
43 + /// Build a new manager.
44 + #[must_use]
45 + pub fn new(store: Arc<dyn GuardStore>, config: GuardConfig) -> Self {
46 + Self { store, config }
47 + }
48 +
49 + /// Borrow the underlying store. Useful for tests and read-only
50 + /// admin paths.
51 + #[must_use]
52 + pub fn store(&self) -> &Arc<dyn GuardStore> {
53 + &self.store
54 + }
55 +
56 + /// Borrow the active [`GuardConfig`].
57 + #[must_use]
58 + pub fn config(&self) -> &GuardConfig {
59 + &self.config
60 + }
61 +
62 + /// Effective decision for `domain` right now.
63 + ///
64 + /// # Errors
65 + ///
66 + /// Propagates errors from the underlying store.
67 + pub async fn decide(&self, domain: &str) -> Result<BanDecision, AppError> {
68 + if !self.config.enabled {
69 + return Ok(BanDecision::Allow);
70 + }
71 + if let Some(acl) = self.store.acl_for_domain(domain).await? {
72 + match acl.kind {
73 + AclKind::Deny => return Ok(BanDecision::Deny),
74 + AclKind::Allow => return Ok(BanDecision::Allow),
75 + }
76 + }
77 + let now = Utc::now();
78 + let active = self.store.active_bans_for_domain(domain, now).await?;
79 + if active.is_empty() {
80 + Ok(BanDecision::Allow)
81 + } else {
82 + Ok(BanDecision::Deny)
83 + }
84 + }
85 +
86 + /// Convenience around [`Self::decide`] returning `true` when the
87 + /// domain is banned.
88 + ///
89 + /// # Errors
90 + ///
91 + /// Propagates errors from the underlying store.
92 + pub async fn is_banned(&self, domain: &str) -> Result<bool, AppError> {
93 + Ok(matches!(self.decide(domain).await?, BanDecision::Deny))
94 + }
95 +
96 + /// Persist a ban triggered by a detector.
97 + ///
98 + /// No-op when the guard is disabled, when `dry_run` is on, when
99 + /// the domain has an explicit allow ACL, or when an equivalent
100 + /// active ban already exists.
101 + ///
102 + /// # Errors
103 + ///
104 + /// Propagates errors from the underlying store.
105 + pub async fn auto_ban(
106 + &self,
107 + domain: &str,
108 + detector_name: &'static str,
109 + reason: impl Into<String>,
110 + ) -> Result<Option<BanEntry>, AppError> {
111 + if !self.config.enabled || self.config.dry_run {
112 + return Ok(None);
113 + }
114 + if let Some(acl) = self.store.acl_for_domain(domain).await? {
115 + if acl.kind == AclKind::Allow {
116 + tracing::debug!(domain, detector_name, "auto-ban skipped: ACL allow");
117 + return Ok(None);
118 + }
119 + }
120 + let now = Utc::now();
121 + if !self.store.active_bans_for_domain(domain, now).await?.is_empty() {
122 + return Ok(None);
123 + }
124 + let expires_at = self.config.auto_ban_duration.map(|d| {
125 + now + chrono::Duration::from_std(d).unwrap_or_else(|_| chrono::Duration::hours(1))
126 + });
127 + let ban = BanEntry {
128 + id: Uuid::new_v4(),
129 + domain: domain.to_string(),
130 + actor_uri: None,
131 + reason: reason.into(),
132 + severity: "suspend".to_string(),
133 + triggered_by: format!("auto:{detector_name}"),
134 + banned_at: now,
135 + expires_at,
136 + lifted_at: None,
137 + lifted_by: None,
138 + };
139 + self.store.insert_ban(&ban).await?;
140 + let evt = GuardEvent {
141 + id: Uuid::new_v4(),
142 + kind: GuardEventKind::Banned,
143 + severity: Severity::Critical,
144 + domain: domain.to_string(),
145 + actor_uri: None,
146 + remote_identity_id: None,
147 + http_status: None,
148 + details: serde_json::json!({
149 + "detector": detector_name,
150 + "ban_id": ban.id,
151 + }),
152 + created_at: now,
153 + };
154 + self.store.insert_event(&evt).await?;
155 + tracing::warn!(
156 + domain,
157 + detector_name,
158 + ban_id = %ban.id,
159 + "fediverse guard auto-ban triggered",
160 + );
161 + Ok(Some(ban))
162 + }
163 +
164 + /// Insert or update an ACL entry.
165 + ///
166 + /// # Errors
167 + ///
168 + /// Propagates errors from the underlying store.
169 + pub async fn upsert_acl(&self, entry: &AclEntry) -> Result<(), AppError> {
170 + self.store.upsert_acl(entry).await
171 + }
172 +
173 + /// Lift a ban manually (admin action).
174 + ///
175 + /// # Errors
176 + ///
177 + /// Propagates errors from the underlying store.
178 + pub async fn lift(&self, ban_id: Uuid, by: Uuid) -> Result<(), AppError> {
179 + self.store.lift_ban(ban_id, Some(by), Utc::now()).await?;
180 + let evt = GuardEvent {
181 + id: Uuid::new_v4(),
182 + kind: GuardEventKind::Unbanned,
183 + severity: Severity::Info,
184 + domain: String::new(), // resolved by reader if needed
185 + actor_uri: None,
186 + remote_identity_id: None,
187 + http_status: None,
188 + details: serde_json::json!({"ban_id": ban_id, "lifted_by": by}),
189 + created_at: Utc::now(),
190 + };
191 + self.store.insert_event(&evt).await
192 + }
193 +}
194 +
195 +#[cfg(test)]
196 +mod tests {
197 + use super::*;
198 + use crate::store::MemoryStore;
199 +
200 + fn manager() -> BanManager {
201 + BanManager::new(Arc::new(MemoryStore::new()), GuardConfig::for_tests())
202 + }
203 +
204 + #[tokio::test]
205 + async fn unknown_domain_is_allowed() {
206 + let m = manager();
207 + assert_eq!(m.decide("clean.tld").await.unwrap(), BanDecision::Allow);
208 + }
209 +
210 + #[tokio::test]
211 + async fn deny_acl_forces_ban() {
212 + let m = manager();
213 + m.upsert_acl(&AclEntry {
214 + id: Uuid::new_v4(),
215 + kind: AclKind::Deny,
216 + domain: "evil.tld".into(),
217 + reason: Some("admin".into()),
218 + created_by: Uuid::new_v4(),
219 + created_at: Utc::now(),
220 + updated_at: Utc::now(),
221 + })
222 + .await
223 + .unwrap();
224 + assert_eq!(m.decide("evil.tld").await.unwrap(), BanDecision::Deny);
225 + }
226 +
227 + #[tokio::test]
228 + async fn allow_acl_blocks_auto_ban() {
229 + let m = manager();
230 + m.upsert_acl(&AclEntry {
231 + id: Uuid::new_v4(),
232 + kind: AclKind::Allow,
233 + domain: "trusted.tld".into(),
234 + reason: Some("peer".into()),
235 + created_by: Uuid::new_v4(),
236 + created_at: Utc::now(),
237 + updated_at: Utc::now(),
238 + })
239 + .await
240 + .unwrap();
241 + let result = m
242 + .auto_ban("trusted.tld", "signature_failure", "test")
243 + .await
244 + .unwrap();
245 + assert!(result.is_none(), "ACL allow must prevent auto-ban");
246 + assert_eq!(m.decide("trusted.tld").await.unwrap(), BanDecision::Allow);
247 + }
248 +
249 + #[tokio::test]
250 + async fn auto_ban_inserts_active_row_and_emits_event() {
251 + let m = manager();
252 + let ban = m
253 + .auto_ban("evil.tld", "signature_failure", "10 sig failures in 5 min")
254 + .await
255 + .unwrap()
256 + .expect("auto_ban must persist a ban");
257 + assert_eq!(ban.domain, "evil.tld");
258 + assert!(ban.triggered_by.starts_with("auto:"));
259 + assert_eq!(m.decide("evil.tld").await.unwrap(), BanDecision::Deny);
260 + }
261 +
262 + #[tokio::test]
263 + async fn auto_ban_is_idempotent_within_window() {
264 + let m = manager();
265 + let first = m.auto_ban("evil.tld", "x", "r").await.unwrap();
266 + let second = m.auto_ban("evil.tld", "x", "r").await.unwrap();
267 + assert!(first.is_some());
268 + assert!(second.is_none(), "second auto_ban must be a no-op");
269 + }
270 +
271 + #[tokio::test]
272 + async fn dry_run_emits_nothing() {
273 + let mut cfg = GuardConfig::for_tests();
274 + cfg.dry_run = true;
275 + let m = BanManager::new(Arc::new(MemoryStore::new()), cfg);
276 + let result = m.auto_ban("evil.tld", "x", "r").await.unwrap();
277 + assert!(result.is_none());
278 + assert_eq!(m.decide("evil.tld").await.unwrap(), BanDecision::Allow);
279 + }
280 +
281 + #[tokio::test]
282 + async fn disabled_guard_always_allows() {
283 + let mut cfg = GuardConfig::for_tests();
284 + cfg.enabled = false;
285 + let m = BanManager::new(Arc::new(MemoryStore::new()), cfg);
286 + // Even with a deny ACL the guard is off.
287 + m.upsert_acl(&AclEntry {
288 + id: Uuid::new_v4(),
289 + kind: AclKind::Deny,
290 + domain: "evil.tld".into(),
291 + reason: None,
292 + created_by: Uuid::new_v4(),
293 + created_at: Utc::now(),
294 + updated_at: Utc::now(),
295 + })
296 + .await
297 + .unwrap();
298 + assert_eq!(m.decide("evil.tld").await.unwrap(), BanDecision::Allow);
299 + }
300 +
301 + #[tokio::test]
302 + async fn lift_marks_ban_inactive_and_emits_unbanned_event() {
303 + let m = manager();
304 + let ban = m
305 + .auto_ban("evil.tld", "x", "r")
306 + .await
307 + .unwrap()
308 + .expect("ban persisted");
309 + m.lift(ban.id, Uuid::new_v4()).await.unwrap();
310 + assert_eq!(m.decide("evil.tld").await.unwrap(), BanDecision::Allow);
311 + }
312 +}
A crates/rustwarden-fediverse-guard/src/config.rs
+205 -0
@@ -0,0 +1,205 @@
1 +//! Runtime configuration for the fediverse guard.
2 +//!
3 +//! Parsed once from `FEDIVERSE_GUARD_*` env vars at boot via
4 +//! [`GuardConfig::from_env`]. Production deployments override defaults
5 +//! through env vars; tests can build a `GuardConfig` literal directly.
6 +
7 +use std::time::Duration;
8 +
9 +use thiserror::Error;
10 +
11 +/// Errors raised while loading [`GuardConfig`] from the environment.
12 +#[derive(Debug, Error)]
13 +pub enum GuardConfigError {
14 + /// An env var was set to a value that could not be parsed.
15 + #[error("invalid value for {name}: {source}")]
16 + InvalidValue {
17 + /// Name of the env var that failed to parse.
18 + name: &'static str,
19 + /// Underlying parser error.
20 + #[source]
21 + source: Box<dyn std::error::Error + Send + Sync>,
22 + },
23 +}
24 +
25 +/// Per-detector tunable threshold. A source (domain or actor) is
26 +/// auto-banned when more than `count` qualifying events are observed
27 +/// within `window`.
28 +#[derive(Debug, Clone, Copy)]
29 +pub struct DetectorThreshold {
30 + /// Maximum events tolerated within the sliding window.
31 + pub count: u32,
32 + /// Sliding-window duration over which `count` is enforced.
33 + pub window: Duration,
34 +}
35 +
36 +impl DetectorThreshold {
37 + /// Build a threshold from explicit values.
38 + #[must_use]
39 + pub const fn new(count: u32, window_secs: u64) -> Self {
40 + Self {
41 + count,
42 + window: Duration::from_secs(window_secs),
43 + }
44 + }
45 +}
46 +
47 +/// Top-level runtime configuration for the fediverse guard.
48 +#[derive(Debug, Clone)]
49 +pub struct GuardConfig {
50 + /// Master kill-switch. When `false`, no detector runs and
51 + /// [`crate::BanManager::is_banned`] always returns `false`.
52 + pub enabled: bool,
53 + /// When `true`, detectors emit events and tracing logs but no ban
54 + /// is ever persisted. Use this to validate thresholds in
55 + /// production before flipping the switch.
56 + pub dry_run: bool,
57 + /// HTTP signatures that fail verification, per signing actor's
58 + /// domain.
59 + pub signature_failures: DetectorThreshold,
60 + /// Detected replays (Date drift / duplicate nonce / signature
61 + /// already seen).
62 + pub replay_attempts: DetectorThreshold,
63 + /// Attempts by the local outbound fetcher to talk to a private/
64 + /// loopback/link-local IP. Surfaces SSRF probing campaigns.
65 + pub fetch_ssrf: DetectorThreshold,
66 + /// Inbox payloads that fail size, schema, or activity-type
67 + /// validation.
68 + pub inbox_abuse: DetectorThreshold,
69 + /// AP-native auth challenges initiated without subsequent consume.
70 + pub auth_spam: DetectorThreshold,
71 + /// TTL applied to auto-bans. `None` means permanent.
72 + pub auto_ban_duration: Option<Duration>,
73 + /// Retention for raw events in `fediverse_guard_events`.
74 + pub events_retention: Duration,
75 +}
76 +
77 +impl GuardConfig {
78 + /// Load configuration from `FEDIVERSE_GUARD_*` environment
79 + /// variables, falling back to documented defaults for any var
80 + /// that is unset.
81 + ///
82 + /// # Errors
83 + ///
84 + /// Returns [`GuardConfigError::InvalidValue`] when an env var
85 + /// is set to a value that cannot be parsed as the expected
86 + /// type.
87 + pub fn from_env() -> Result<Self, GuardConfigError> {
88 + Ok(Self {
89 + enabled: parse_bool_env("FEDIVERSE_GUARD_ENABLED", true)?,
90 + dry_run: parse_bool_env("FEDIVERSE_GUARD_DRY_RUN", false)?,
91 + signature_failures: DetectorThreshold {
92 + count: parse_u32_env("FEDIVERSE_GUARD_SIG_FAILURE_THRESHOLD", 10)?,
93 + window: Duration::from_secs(parse_u64_env(
94 + "FEDIVERSE_GUARD_SIG_FAILURE_WINDOW_SECS",
95 + 300,
96 + )?),
97 + },
98 + replay_attempts: DetectorThreshold {
99 + count: parse_u32_env("FEDIVERSE_GUARD_REPLAY_THRESHOLD", 3)?,
100 + window: Duration::from_secs(parse_u64_env(
101 + "FEDIVERSE_GUARD_REPLAY_WINDOW_SECS",
102 + 600,
103 + )?),
104 + },
105 + fetch_ssrf: DetectorThreshold {
106 + count: parse_u32_env("FEDIVERSE_GUARD_SSRF_THRESHOLD", 5)?,
107 + window: Duration::from_secs(parse_u64_env(
108 + "FEDIVERSE_GUARD_SSRF_WINDOW_SECS",
109 + 600,
110 + )?),
111 + },
112 + inbox_abuse: DetectorThreshold {
113 + count: parse_u32_env("FEDIVERSE_GUARD_INBOX_ABUSE_THRESHOLD", 20)?,
114 + window: Duration::from_secs(parse_u64_env(
115 + "FEDIVERSE_GUARD_INBOX_ABUSE_WINDOW_SECS",
116 + 300,
117 + )?),
118 + },
119 + auth_spam: DetectorThreshold {
120 + count: parse_u32_env("FEDIVERSE_GUARD_AUTH_SPAM_THRESHOLD", 30)?,
121 + window: Duration::from_secs(parse_u64_env(
122 + "FEDIVERSE_GUARD_AUTH_SPAM_WINDOW_SECS",
123 + 300,
124 + )?),
125 + },
126 + auto_ban_duration: {
127 + let secs = parse_u64_env("FEDIVERSE_GUARD_AUTO_BAN_DURATION_SECS", 3600)?;
128 + if secs == 0 { None } else { Some(Duration::from_secs(secs)) }
129 + },
130 + events_retention: Duration::from_secs(
131 + parse_u64_env("FEDIVERSE_GUARD_EVENTS_RETENTION_DAYS", 90)? * 24 * 3600,
132 + ),
133 + })
134 + }
135 +
136 + /// Build a permissive `GuardConfig` for tests.
137 + #[must_use]
138 + pub fn for_tests() -> Self {
139 + Self {
140 + enabled: true,
141 + dry_run: false,
142 + signature_failures: DetectorThreshold::new(3, 60),
143 + replay_attempts: DetectorThreshold::new(2, 60),
144 + fetch_ssrf: DetectorThreshold::new(3, 60),
145 + inbox_abuse: DetectorThreshold::new(5, 60),
146 + auth_spam: DetectorThreshold::new(5, 60),
147 + auto_ban_duration: Some(Duration::from_secs(60)),
148 + events_retention: Duration::from_secs(3600),
149 + }
150 + }
151 +}
152 +
153 +fn parse_bool_env(name: &'static str, default: bool) -> Result<bool, GuardConfigError> {
154 + match std::env::var(name) {
155 + Ok(v) => match v.to_ascii_lowercase().as_str() {
156 + "true" | "1" | "yes" => Ok(true),
157 + "false" | "0" | "no" => Ok(false),
158 + other => Err(GuardConfigError::InvalidValue {
159 + name,
160 + source: format!("expected true/false, got {other:?}").into(),
161 + }),
162 + },
163 + Err(_) => Ok(default),
164 + }
165 +}
166 +
167 +fn parse_u32_env(name: &'static str, default: u32) -> Result<u32, GuardConfigError> {
168 + match std::env::var(name) {
169 + Ok(v) => v.parse::<u32>().map_err(|e| GuardConfigError::InvalidValue {
170 + name,
171 + source: Box::new(e),
172 + }),
173 + Err(_) => Ok(default),
174 + }
175 +}
176 +
177 +fn parse_u64_env(name: &'static str, default: u64) -> Result<u64, GuardConfigError> {
178 + match std::env::var(name) {
179 + Ok(v) => v.parse::<u64>().map_err(|e| GuardConfigError::InvalidValue {
180 + name,
181 + source: Box::new(e),
182 + }),
183 + Err(_) => Ok(default),
184 + }
185 +}
186 +
187 +#[cfg(test)]
188 +mod tests {
189 + use super::*;
190 +
191 + #[test]
192 + fn for_tests_yields_sane_defaults() {
193 + let cfg = GuardConfig::for_tests();
194 + assert!(cfg.enabled);
195 + assert!(!cfg.dry_run);
196 + assert_eq!(cfg.signature_failures.count, 3);
197 + }
198 +
199 + #[test]
200 + fn detector_threshold_new_converts_secs_to_duration() {
201 + let t = DetectorThreshold::new(7, 120);
202 + assert_eq!(t.count, 7);
203 + assert_eq!(t.window, Duration::from_secs(120));
204 + }
205 +}
A crates/rustwarden-fediverse-guard/src/events.rs
+141 -0
@@ -0,0 +1,141 @@
1 +//! Stable on-the-wire format for guard events plus the [`GuardEventSink`]
2 +//! trait that backends (tracing, file, fan-out) implement.
3 +
4 +use async_trait::async_trait;
5 +use chrono::{DateTime, Utc};
6 +use serde::{Deserialize, Serialize};
7 +use uuid::Uuid;
8 +
9 +/// Discriminator describing what an observer reports.
10 +///
11 +/// Each variant corresponds to one of the five detectors. Variants are
12 +/// serialised in `snake_case` to remain stable in JSON logs consumed
13 +/// by external SIEMs.
14 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
15 +#[serde(rename_all = "snake_case")]
16 +pub enum GuardEventKind {
17 + /// A request's HTTP signature failed cryptographic verification.
18 + SignatureFailure,
19 + /// Replay window violated: stale `Date` header or duplicate nonce.
20 + ReplayAttempt,
21 + /// Outbound fetcher attempted to reach a private/loopback/link-local
22 + /// IP (SSRF guard hit).
23 + FetchSsrf,
24 + /// Inbox payload failed size, schema, or type validation.
25 + InboxAbuse,
26 + /// AP-native auth challenge initiated but never consumed; high
27 + /// volume from one domain signals scripted abuse.
28 + AuthSpam,
29 + /// A domain or actor was banned (auto or manual). Reported by the
30 + /// `BanManager` itself, not by detectors.
31 + Banned,
32 + /// A domain or actor ban was lifted (TTL expiry or admin action).
33 + Unbanned,
34 +}
35 +
36 +/// Severity hint attached to each event. Mirrors the levels accepted
37 +/// by `fediverse_guard_events.severity`.
38 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
39 +#[serde(rename_all = "lowercase")]
40 +pub enum Severity {
41 + /// Informational (e.g. ban lifted by TTL).
42 + Info,
43 + /// Low — anomaly worth recording but not actionable alone.
44 + Low,
45 + /// Medium — repeated occurrences should trigger a review.
46 + Medium,
47 + /// High — likely abuse; detector seuils probably reached soon.
48 + High,
49 + /// Critical — confirmed abuse; auto-ban just triggered.
50 + Critical,
51 +}
52 +
53 +/// One observable guard event.
54 +///
55 +/// Events are produced by the federation crate via the
56 +/// `FediverseEventObserver` trait and consumed by the guard's detector
57 +/// chain and sinks.
58 +#[derive(Debug, Clone, Serialize, Deserialize)]
59 +pub struct GuardEvent {
60 + /// Surrogate UUID for the event row.
61 + pub id: Uuid,
62 + /// What kind of anomaly is being reported.
63 + pub kind: GuardEventKind,
64 + /// Severity hint.
65 + pub severity: Severity,
66 + /// Domain the event is attributed to (lowercased, no scheme).
67 + pub domain: String,
68 + /// Optional ActivityPub actor URI (e.g.
69 + /// `https://a.tld/users/alice`) when the event narrows to one actor.
70 + pub actor_uri: Option<String>,
71 + /// FK to `remote_identities.id` when the actor has been resolved.
72 + pub remote_identity_id: Option<Uuid>,
73 + /// Optional HTTP status returned by the offending request, when
74 + /// applicable.
75 + pub http_status: Option<i32>,
76 + /// Detector-specific metadata (e.g. `{"skew_secs": 3601}`).
77 + /// Always a JSON object; never `null`.
78 + pub details: serde_json::Value,
79 + /// Server-side timestamp.
80 + pub created_at: DateTime<Utc>,
81 +}
82 +
83 +impl GuardEvent {
84 + /// Convenience constructor with sensible defaults for fields
85 + /// rarely populated by detectors.
86 + #[must_use]
87 + pub fn new(kind: GuardEventKind, domain: impl Into<String>, severity: Severity) -> Self {
88 + Self {
89 + id: Uuid::new_v4(),
90 + kind,
91 + severity,
92 + domain: domain.into(),
93 + actor_uri: None,
94 + remote_identity_id: None,
95 + http_status: None,
96 + details: serde_json::Value::Object(serde_json::Map::new()),
97 + created_at: Utc::now(),
98 + }
99 + }
100 +}
101 +
102 +/// Pluggable sink for [`GuardEvent`].
103 +///
104 +/// Implementations must be non-blocking: a slow sink should buffer
105 +/// internally rather than back-pressure the federation hot path.
106 +#[async_trait]
107 +pub trait GuardEventSink: Send + Sync {
108 + /// Emit one event. Errors are swallowed by the caller — sinks
109 + /// must not propagate failures into the request path.
110 + async fn emit(&self, event: &GuardEvent);
111 +}
112 +
113 +#[cfg(test)]
114 +mod tests {
115 + use super::*;
116 +
117 + #[test]
118 + fn guard_event_new_fills_defaults() {
119 + let evt = GuardEvent::new(GuardEventKind::SignatureFailure, "evil.tld", Severity::High);
120 + assert_eq!(evt.kind, GuardEventKind::SignatureFailure);
121 + assert_eq!(evt.domain, "evil.tld");
122 + assert!(evt.actor_uri.is_none());
123 + assert!(evt.http_status.is_none());
124 + assert!(evt.details.is_object());
125 + }
126 +
127 + #[test]
128 + fn guard_event_round_trips_through_json() {
129 + let evt = GuardEvent::new(GuardEventKind::ReplayAttempt, "a.tld", Severity::Medium);
130 + let json = serde_json::to_string(&evt).unwrap();
131 + let back: GuardEvent = serde_json::from_str(&json).unwrap();
132 + assert_eq!(back.kind, GuardEventKind::ReplayAttempt);
133 + assert_eq!(back.domain, "a.tld");
134 + }
135 +
136 + #[test]
137 + fn guard_event_kind_serialises_snake_case() {
138 + let s = serde_json::to_string(&GuardEventKind::SignatureFailure).unwrap();
139 + assert_eq!(s, "\"signature_failure\"");
140 + }
141 +}
A crates/rustwarden-fediverse-guard/src/lib.rs
+30 -0
@@ -0,0 +1,30 @@
1 +//! Real-time abuse detection and ban management for the Rustwarden
2 +//! ActivityPub federation layer (RFC 0028+).
3 +//!
4 +//! Mirrors the pattern of `gitrust-ssh-guard` adapted to ActivityPub
5 +//! semantics: domain- and actor-level events feed a sliding-window
6 +//! detector chain that may auto-ban abusive sources. Admin overrides
7 +//! and a manual allow/deny ACL sit on top, all backed by a hybrid
8 +//! in-memory plus Postgres store for hot-path checks at request time.
9 +//!
10 +//! # Architecture
11 +//!
12 +//! ```text
13 +//! incoming AP event ───► GuardEvent ───► sinks (tracing/file)
14 +//! └──► store.insert_event()
15 +//! └──► detectors (per-domain windows)
16 +//! └► BanManager::auto_ban()
17 +//! └► store.insert_ban()
18 +//! request hot path ──► BanManager::is_banned(domain) ──► allow / deny
19 +//! ```
20 +
21 +pub mod ban;
22 +pub mod config;
23 +pub mod events;
24 +pub mod sinks;
25 +pub mod store;
26 +
27 +pub use ban::{BanDecision, BanManager};
28 +pub use config::{DetectorThreshold, GuardConfig, GuardConfigError};
29 +pub use events::{GuardEvent, GuardEventKind, GuardEventSink};
30 +pub use store::{AclEntry, AclKind, BanEntry, GuardStore, MemoryStore};
A crates/rustwarden-fediverse-guard/src/sinks.rs
+98 -0
@@ -0,0 +1,98 @@
1 +//! Built-in [`GuardEventSink`] implementations.
2 +//!
3 +//! Two sinks are provided out of the box:
4 +//!
5 +//! - [`TracingSink`] — emits each event via `tracing::warn!` under
6 +//! the `rustwarden_fediverse_guard` target, picked up by journald,
7 +//! `tracing_subscriber` console layers, or OpenTelemetry exporters.
8 +//! - [`MultiSink`] — fan-out to several sinks sequentially. A slow
9 +//! sink does **not** back-pressure the federation hot path because
10 +//! `emit` is fire-and-forget on each.
11 +
12 +use std::sync::Arc;
13 +
14 +use async_trait::async_trait;
15 +
16 +use crate::events::{GuardEvent, GuardEventSink};
17 +
18 +/// Sink that emits each event via `tracing::warn!`.
19 +#[derive(Debug, Default, Clone, Copy)]
20 +pub struct TracingSink;
21 +
22 +impl TracingSink {
23 + /// Build a new tracing sink.
24 + #[must_use]
25 + pub const fn new() -> Self {
26 + Self
27 + }
28 +}
29 +
30 +#[async_trait]
31 +impl GuardEventSink for TracingSink {
32 + async fn emit(&self, event: &GuardEvent) {
33 + tracing::warn!(
34 + target: "rustwarden_fediverse_guard",
35 + event_kind = ?event.kind,
36 + severity = ?event.severity,
37 + domain = %event.domain,
38 + actor_uri = event.actor_uri.as_deref().unwrap_or(""),
39 + details = %event.details,
40 + "fediverse guard event",
41 + );
42 + }
43 +}
44 +
45 +/// Fan-out sink that calls every child in sequence.
46 +pub struct MultiSink {
47 + children: Vec<Arc<dyn GuardEventSink>>,
48 +}
49 +
50 +impl MultiSink {
51 + /// Build a new multi-sink from a vector of children.
52 + #[must_use]
53 + pub fn new(children: Vec<Arc<dyn GuardEventSink>>) -> Self {
54 + Self { children }
55 + }
56 +}
57 +
58 +#[async_trait]
59 +impl GuardEventSink for MultiSink {
60 + async fn emit(&self, event: &GuardEvent) {
61 + for child in &self.children {
62 + child.emit(event).await;
63 + }
64 + }
65 +}
66 +
67 +#[cfg(test)]
68 +mod tests {
69 + use super::*;
70 + use crate::events::{GuardEventKind, Severity};
71 + use std::sync::Mutex;
72 +
73 + struct Recorder {
74 + events: Mutex<Vec<GuardEvent>>,
75 + }
76 +
77 + #[async_trait]
78 + impl GuardEventSink for Recorder {
79 + async fn emit(&self, event: &GuardEvent) {
80 + self.events.lock().unwrap().push(event.clone());
81 + }
82 + }
83 +
84 + #[tokio::test]
85 + async fn multi_sink_dispatches_to_every_child() {
86 + let r1 = Arc::new(Recorder {
87 + events: Mutex::new(Vec::new()),
88 + });
89 + let r2 = Arc::new(Recorder {
90 + events: Mutex::new(Vec::new()),
91 + });
92 + let multi = MultiSink::new(vec![r1.clone() as Arc<dyn GuardEventSink>, r2.clone()]);
93 + let evt = GuardEvent::new(GuardEventKind::SignatureFailure, "x.tld", Severity::Medium);
94 + multi.emit(&evt).await;
95 + assert_eq!(r1.events.lock().unwrap().len(), 1);
96 + assert_eq!(r2.events.lock().unwrap().len(), 1);
97 + }
98 +}
A crates/rustwarden-fediverse-guard/src/store/memory.rs
+187 -0
@@ -0,0 +1,187 @@
1 +//! In-process [`GuardStore`] backend.
2 +//!
3 +//! Implemented with `DashMap` for O(1) hot-path reads. Events, bans,
4 +//! and ACLs live only in RAM — restarting the process clears
5 +//! everything. Use this in tests, or wrapped by a write-through
6 +//! `HybridStore` in production.
7 +
8 +use async_trait::async_trait;
9 +use chrono::{DateTime, Utc};
10 +use dashmap::DashMap;
11 +use rustwarden_core::error::AppError;
12 +use uuid::Uuid;
13 +
14 +use super::{AclEntry, BanEntry, GuardStore};
15 +use crate::events::{GuardEvent, GuardEventKind};
16 +
17 +/// Concurrent in-memory store.
18 +#[derive(Default)]
19 +pub struct MemoryStore {
20 + events: DashMap<String, Vec<GuardEvent>>,
21 + bans: DashMap<Uuid, BanEntry>,
22 + acl: DashMap<String, AclEntry>,
23 +}
24 +
25 +impl MemoryStore {
26 + /// Build an empty store.
27 + #[must_use]
28 + pub fn new() -> Self {
29 + Self::default()
30 + }
31 +}
32 +
33 +#[async_trait]
34 +impl GuardStore for MemoryStore {
35 + async fn insert_event(&self, event: &GuardEvent) -> Result<(), AppError> {
36 + self.events
37 + .entry(event.domain.clone())
38 + .or_default()
39 + .push(event.clone());
40 + Ok(())
41 + }
42 +
43 + async fn count_events_since(
44 + &self,
45 + domain: &str,
46 + kind: GuardEventKind,
47 + since: DateTime<Utc>,
48 + ) -> Result<u64, AppError> {
49 + let count = self
50 + .events
51 + .get(domain)
52 + .map(|bucket| {
53 + bucket
54 + .iter()
55 + .filter(|e| e.kind == kind && e.created_at >= since)
56 + .count() as u64
57 + })
58 + .unwrap_or(0);
59 + Ok(count)
60 + }
61 +
62 + async fn insert_ban(&self, ban: &BanEntry) -> Result<(), AppError> {
63 + self.bans.insert(ban.id, ban.clone());
64 + Ok(())
65 + }
66 +
67 + async fn active_bans_for_domain(
68 + &self,
69 + domain: &str,
70 + now: DateTime<Utc>,
71 + ) -> Result<Vec<BanEntry>, AppError> {
72 + let out = self
73 + .bans
74 + .iter()
75 + .filter(|kv| kv.value().domain == domain && kv.value().is_active_at(now))
76 + .map(|kv| kv.value().clone())
77 + .collect();
78 + Ok(out)
79 + }
80 +
81 + async fn lift_ban(
82 + &self,
83 + ban_id: Uuid,
84 + by: Option<Uuid>,
85 + at: DateTime<Utc>,
86 + ) -> Result<(), AppError> {
87 + if let Some(mut row) = self.bans.get_mut(&ban_id) {
88 + row.lifted_at = Some(at);
89 + row.lifted_by = by;
90 + }
91 + Ok(())
92 + }
93 +
94 + async fn upsert_acl(&self, entry: &AclEntry) -> Result<(), AppError> {
95 + self.acl.insert(entry.domain.clone(), entry.clone());
96 + Ok(())
97 + }
98 +
99 + async fn acl_for_domain(&self, domain: &str) -> Result<Option<AclEntry>, AppError> {
100 + Ok(self.acl.get(domain).map(|kv| kv.value().clone()))
101 + }
102 +
103 + async fn delete_acl(&self, id: Uuid) -> Result<(), AppError> {
104 + self.acl.retain(|_, v| v.id != id);
105 + Ok(())
106 + }
107 +}
108 +
109 +#[cfg(test)]
110 +mod tests {
111 + use super::*;
112 + use crate::events::{GuardEvent, GuardEventKind, Severity};
113 + use crate::store::AclKind;
114 +
115 + fn evt(domain: &str, kind: GuardEventKind) -> GuardEvent {
116 + GuardEvent::new(kind, domain, Severity::Medium)
117 + }
118 +
119 + #[tokio::test]
120 + async fn insert_and_count_events_by_kind_and_window() {
121 + let store = MemoryStore::new();
122 + let e1 = evt("a.tld", GuardEventKind::SignatureFailure);
123 + let e2 = evt("a.tld", GuardEventKind::SignatureFailure);
124 + let e3 = evt("a.tld", GuardEventKind::ReplayAttempt);
125 + store.insert_event(&e1).await.unwrap();
126 + store.insert_event(&e2).await.unwrap();
127 + store.insert_event(&e3).await.unwrap();
128 + let cutoff = Utc::now() - chrono::Duration::minutes(1);
129 + let n = store
130 + .count_events_since("a.tld", GuardEventKind::SignatureFailure, cutoff)
131 + .await
132 + .unwrap();
133 + assert_eq!(n, 2);
134 + }
135 +
136 + #[tokio::test]
137 + async fn ban_lifecycle_insert_active_lift() {
138 + let store = MemoryStore::new();
139 + let now = Utc::now();
140 + let ban = BanEntry {
141 + id: Uuid::new_v4(),
142 + domain: "evil.tld".into(),
143 + actor_uri: None,
144 + reason: "auto-ban".into(),
145 + severity: "suspend".into(),
146 + triggered_by: "auto:test".into(),
147 + banned_at: now,
148 + expires_at: Some(now + chrono::Duration::hours(1)),
149 + lifted_at: None,
150 + lifted_by: None,
151 + };
152 + let ban_id = ban.id;
153 + store.insert_ban(&ban).await.unwrap();
154 + let active = store
155 + .active_bans_for_domain("evil.tld", now + chrono::Duration::minutes(10))
156 + .await
157 + .unwrap();
158 + assert_eq!(active.len(), 1);
159 + let admin = Uuid::new_v4();
160 + store.lift_ban(ban_id, Some(admin), Utc::now()).await.unwrap();
161 + let active_after = store
162 + .active_bans_for_domain("evil.tld", now + chrono::Duration::minutes(10))
163 + .await
164 + .unwrap();
165 + assert!(active_after.is_empty());
166 + }
167 +
168 + #[tokio::test]
169 + async fn acl_upsert_lookup_delete() {
170 + let store = MemoryStore::new();
171 + let entry = AclEntry {
172 + id: Uuid::new_v4(),
173 + kind: AclKind::Allow,
174 + domain: "trusted.tld".into(),
175 + reason: Some("peer".into()),
176 + created_by: Uuid::new_v4(),
177 + created_at: Utc::now(),
178 + updated_at: Utc::now(),
179 + };
180 + store.upsert_acl(&entry).await.unwrap();
181 + let found = store.acl_for_domain("trusted.tld").await.unwrap();
182 + assert!(found.is_some());
183 + store.delete_acl(entry.id).await.unwrap();
184 + let after = store.acl_for_domain("trusted.tld").await.unwrap();
185 + assert!(after.is_none());
186 + }
187 +}
A crates/rustwarden-fediverse-guard/src/store/mod.rs
+202 -0
@@ -0,0 +1,202 @@
1 +//! Persistence layer for the fediverse guard.
2 +//!
3 +//! Three interchangeable backends implement the [`GuardStore`] trait:
4 +//!
5 +//! - [`MemoryStore`] — in-process `DashMap`, O(1) hot-path, lost on restart.
6 +//! - `PostgresStore` (forthcoming) — SeaORM-backed durable store.
7 +//! - `HybridStore` (forthcoming) — memory cache + write-through Postgres
8 +//! plus boot-time rehydration.
9 +
10 +mod memory;
11 +
12 +use async_trait::async_trait;
13 +use chrono::{DateTime, Utc};
14 +use rustwarden_core::error::AppError;
15 +use serde::{Deserialize, Serialize};
16 +use uuid::Uuid;
17 +
18 +pub use memory::MemoryStore;
19 +
20 +use crate::events::{GuardEvent, GuardEventKind};
21 +
22 +/// Active ban row.
23 +#[derive(Debug, Clone, Serialize, Deserialize)]
24 +pub struct BanEntry {
25 + /// Surrogate UUID for the ban row.
26 + pub id: Uuid,
27 + /// Lower-cased domain the ban applies to (always populated).
28 + pub domain: String,
29 + /// Optional ActivityPub actor URI to narrow the ban to a single
30 + /// account on `domain`. `None` means the whole domain is banned.
31 + pub actor_uri: Option<String>,
32 + /// Human-readable reason recorded for the audit trail.
33 + pub reason: String,
34 + /// Severity: `"suspend"`, `"silence"`, or `"rate_limit"`.
35 + pub severity: String,
36 + /// `"auto:<detector_name>"` for detector-driven bans, or
37 + /// `"manual:<admin_uuid>"` for an admin's manual action.
38 + pub triggered_by: String,
39 + /// When the ban was first applied.
40 + pub banned_at: DateTime<Utc>,
41 + /// `None` means permanent.
42 + pub expires_at: Option<DateTime<Utc>>,
43 + /// Set when the ban was lifted (TTL or admin override).
44 + pub lifted_at: Option<DateTime<Utc>>,
45 + /// Admin user that lifted the ban, when applicable.
46 + pub lifted_by: Option<Uuid>,
47 +}
48 +
49 +impl BanEntry {
50 + /// Returns `true` if this row is still active at `now` (i.e. not
51 + /// lifted and either permanent or before `expires_at`).
52 + #[must_use]
53 + pub fn is_active_at(&self, now: DateTime<Utc>) -> bool {
54 + if self.lifted_at.is_some() {
55 + return false;
56 + }
57 + match self.expires_at {
58 + Some(exp) => now < exp,
59 + None => true,
60 + }
61 + }
62 +}
63 +
64 +/// Manual allow/deny list entry. Allow rules bypass auto-ban; deny
65 +/// rules force ban even before a detector trips.
66 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
67 +#[serde(rename_all = "lowercase")]
68 +pub enum AclKind {
69 + /// The domain is whitelisted; auto-ban detectors must not target it.
70 + Allow,
71 + /// The domain is blacklisted; treated as permanently banned.
72 + Deny,
73 +}
74 +
75 +/// Manual ACL row.
76 +#[derive(Debug, Clone, Serialize, Deserialize)]
77 +pub struct AclEntry {
78 + /// Surrogate UUID for the ACL row.
79 + pub id: Uuid,
80 + /// Allow or deny.
81 + pub kind: AclKind,
82 + /// Lower-cased domain the entry applies to.
83 + pub domain: String,
84 + /// Free-form reason captured for audit.
85 + pub reason: Option<String>,
86 + /// FK to `users.id` of the admin that created the entry.
87 + pub created_by: Uuid,
88 + /// When the entry was first inserted.
89 + pub created_at: DateTime<Utc>,
90 + /// Last mutation timestamp.
91 + pub updated_at: DateTime<Utc>,
92 +}
93 +
94 +/// Abstract backend for guard state.
95 +///
96 +/// Implementations must be cheap to clone (`Arc`) and safe to share
97 +/// across handlers and background workers.
98 +#[async_trait]
99 +pub trait GuardStore: Send + Sync {
100 + /// Persist one raw event.
101 + async fn insert_event(&self, event: &GuardEvent) -> Result<(), AppError>;
102 +
103 + /// Count events of `kind` for `domain` since `since`. Powers the
104 + /// sliding-window detectors.
105 + async fn count_events_since(
106 + &self,
107 + domain: &str,
108 + kind: GuardEventKind,
109 + since: DateTime<Utc>,
110 + ) -> Result<u64, AppError>;
111 +
112 + /// Insert a new ban row. Implementations may decline silently if
113 + /// an equivalent active ban already exists.
114 + async fn insert_ban(&self, ban: &BanEntry) -> Result<(), AppError>;
115 +
116 + /// Return every active ban whose domain matches `domain` and that
117 + /// is still effective at `now`.
118 + async fn active_bans_for_domain(
119 + &self,
120 + domain: &str,
121 + now: DateTime<Utc>,
122 + ) -> Result<Vec<BanEntry>, AppError>;
123 +
124 + /// Mark a ban as lifted.
125 + async fn lift_ban(
126 + &self,
127 + ban_id: Uuid,
128 + by: Option<Uuid>,
129 + at: DateTime<Utc>,
130 + ) -> Result<(), AppError>;
131 +
132 + /// Insert or update an ACL row.
133 + async fn upsert_acl(&self, entry: &AclEntry) -> Result<(), AppError>;
134 +
135 + /// Look up the ACL kind for `domain`, if any.
136 + async fn acl_for_domain(&self, domain: &str) -> Result<Option<AclEntry>, AppError>;
137 +
138 + /// Remove an ACL row by id.
139 + async fn delete_acl(&self, id: Uuid) -> Result<(), AppError>;
140 +}
141 +
142 +/// Normalise a user-supplied domain string to the canonical form used
143 +/// everywhere in this crate (lowercase, stripped of scheme/trailing
144 +/// slash/whitespace).
145 +#[must_use]
146 +pub fn normalise_domain(input: &str) -> String {
147 + input
148 + .trim()
149 + .trim_start_matches("https://")
150 + .trim_start_matches("http://")
151 + .trim_end_matches('/')
152 + .to_ascii_lowercase()
153 +}
154 +
155 +#[cfg(test)]
156 +mod tests {
157 + use super::*;
158 +
159 + #[test]
160 + fn normalise_domain_strips_scheme_and_trailing_slash() {
161 + assert_eq!(normalise_domain(" https://Foo.TLD/ "), "foo.tld");
162 + assert_eq!(normalise_domain("http://bar.org"), "bar.org");
163 + assert_eq!(normalise_domain("baz.net"), "baz.net");
164 + }
165 +
166 + #[test]
167 + fn ban_entry_is_active_until_expires_at() {
168 + let now = Utc::now();
169 + let active = BanEntry {
170 + id: Uuid::new_v4(),
171 + domain: "evil.tld".into(),
172 + actor_uri: None,
173 + reason: "spam".into(),
174 + severity: "suspend".into(),
175 + triggered_by: "auto:signature_failure".into(),
176 + banned_at: now - chrono::Duration::minutes(1),
177 + expires_at: Some(now + chrono::Duration::hours(1)),
178 + lifted_at: None,
179 + lifted_by: None,
180 + };
181 + assert!(active.is_active_at(now));
182 + assert!(!active.is_active_at(now + chrono::Duration::hours(2)));
183 + }
184 +
185 + #[test]
186 + fn ban_entry_lifted_is_inactive_even_before_expiry() {
187 + let now = Utc::now();
188 + let lifted = BanEntry {
189 + id: Uuid::new_v4(),
190 + domain: "evil.tld".into(),
191 + actor_uri: None,
192 + reason: "spam".into(),
193 + severity: "suspend".into(),
194 + triggered_by: "manual:admin".into(),
195 + banned_at: now - chrono::Duration::minutes(10),
196 + expires_at: Some(now + chrono::Duration::hours(1)),
197 + lifted_at: Some(now - chrono::Duration::minutes(1)),
198 + lifted_by: Some(Uuid::new_v4()),
199 + };
200 + assert!(!lifted.is_active_at(now));
201 + }
202 +}