From 07ddfd0ddcf6c96ef875a23cc52f96608cc77669 Mon Sep 17 00:00:00 2001 From: usman222nuhu Date: Fri, 16 Oct 2026 05:50:13 +0000 Subject: [PATCH 1/2] feat(api-server): add GDPR user data deletion endpoint (#1107) Adds DELETE /users/{user_id}, which erases a user's off-chain data: - commitment index entries and tags for IPs they own, cascading to every watchlist that referenced them - the user's own watchlist, 2FA configuration and sessions - audit log events attributed to the user, anonymized with the HMAC chain re-signed so verification still passes - post-erasure scan that reports any remaining references On-chain commitments are immutable and are reported as retained. --- api-server/src/audit.rs | 160 ++++++++++++++++++ api-server/src/auth_2fa.rs | 6 + api-server/src/commitments.rs | 5 + api-server/src/handlers.rs | 12 +- api-server/src/main.rs | 2 + api-server/src/session.rs | 14 ++ api-server/src/user_data.rs | 301 ++++++++++++++++++++++++++++++++++ 7 files changed, 494 insertions(+), 6 deletions(-) create mode 100644 api-server/src/user_data.rs diff --git a/api-server/src/audit.rs b/api-server/src/audit.rs index 72ad74a6..d05a0aa6 100644 --- a/api-server/src/audit.rs +++ b/api-server/src/audit.rs @@ -436,6 +436,41 @@ impl AuditLogStore { suspicious } + + /// Erases personal data recorded against `actor_hash` (#1107). + /// + /// Matching events keep their sequence, timestamp, category and outcome, + /// but lose `actor_hash`, `request_id`, `trace_id` and `details`. Each + /// signature covers the previous one, so the chain from the first affected + /// event onward is re-signed with the server key and rewritten atomically. + /// Returns the number of events anonymized. + pub async fn anonymize_actor(&self, actor_hash: &str) -> Result { + let key = self.key.clone(); + let path = self.path.clone(); + let target = actor_hash.to_string(); + + let (count, rewritten) = + tokio::task::spawn_blocking(move || anonymize_actor_locked(&path, &key, &target)) + .await + .map_err(|error| AuditLogError::Io(format!("anonymize task panicked: {error}")))??; + + if count > 0 { + *self.events.write().await = rewritten; + } + + Ok(count) + } + + /// Counts persisted events still attributed to `actor_hash`. Used to + /// verify that an erasure left no references behind. + pub async fn count_actor_events(&self, actor_hash: &str) -> usize { + self.events + .read() + .await + .iter() + .filter(|event| event.actor_hash.as_deref() == Some(actor_hash)) + .count() + } } /// Walks the persisted chain at `path` and confirms every `signature` @@ -648,6 +683,65 @@ fn append_locked( Ok(event) } +/// Rewrites the backing file with personal data for `target` removed and the +/// chain re-signed from the first affected event. Runs under the append lock +/// so no concurrent append can interleave with the rewrite. +fn anonymize_actor_locked( + path: &Path, + key: &[u8], + target: &str, +) -> Result<(usize, Vec), AuditLogError> { + let _lock = AppendLockGuard::acquire(path)?; + + let mut events = read_events(path)?; + let mut count = 0; + let mut first_changed: Option = None; + + for (index, event) in events.iter_mut().enumerate() { + if event.actor_hash.as_deref() == Some(target) { + event.actor_hash = None; + event.request_id = None; + event.trace_id = None; + event.details = Value::Object(Map::new()); + count += 1; + first_changed.get_or_insert(index); + } + } + + let Some(start) = first_changed else { + return Ok((0, events)); + }; + + let mut previous_signature = if start == 0 { + GENESIS_SIGNATURE.to_string() + } else { + events[start - 1].signature.clone() + }; + for event in events.iter_mut().skip(start) { + event.previous_signature = previous_signature.clone(); + event.signature = compute_signature(key, event); + previous_signature = event.signature.clone(); + } + + let mut body = String::new(); + for event in &events { + body.push_str(&serde_json::to_string(event).map_err(|_| AuditLogError::Serialization)?); + body.push('\n'); + } + + let mut tmp_os = path.as_os_str().to_owned(); + tmp_os.push(".tmp"); + let tmp_path = PathBuf::from(tmp_os); + { + let mut file = File::create(&tmp_path)?; + file.write_all(body.as_bytes())?; + file.sync_all()?; + } + fs::rename(&tmp_path, path)?; + + Ok((count, events)) +} + /// Reads and parses every record in the backing file. A trailing line that /// fails to parse is treated as an interrupted write (the process crashed /// between `write_all` and `sync_all`) and is dropped rather than rejected, @@ -972,4 +1066,70 @@ mod tests { assert_eq!(dirty.status, AuditIntegrityStatus::Invalid); assert_eq!(dirty.broken_at_sequence, Some(1)); } + + #[tokio::test] + async fn anonymize_actor_redacts_personal_data_and_keeps_chain_valid() { + let key = vec![9; 32]; + let (store, temp) = open_store("anonymize", key.clone()); + let target = store.hash_identifier("user-to-erase"); + let other = store.hash_identifier("bystander"); + + store + .append( + AuditEventInput::new(AuditEventCategory::Commitment, "ip.commit.requested") + .actor_hash(Some(target.clone())) + .request_id(Some("req-1".to_string())) + .detail("owner", "user-to-erase"), + ) + .await + .unwrap(); + store + .append( + AuditEventInput::new(AuditEventCategory::Swap, "swap.accept.requested") + .actor_hash(Some(other.clone())) + .detail("buyer", "bystander"), + ) + .await + .unwrap(); + store + .append( + AuditEventInput::new(AuditEventCategory::Admin, "admin.action") + .actor_hash(Some(target.clone())), + ) + .await + .unwrap(); + + let count = store.anonymize_actor(&target).await.unwrap(); + + assert_eq!(count, 2); + assert_eq!(store.count_actor_events(&target).await, 0); + assert_eq!(store.count_actor_events(&other).await, 1); + + let persisted = read_events(temp.path()).unwrap(); + assert_eq!(persisted.len(), 3); + assert_eq!(persisted[0].actor_hash, None); + assert_eq!(persisted[0].request_id, None); + assert_eq!(persisted[0].details, Value::Object(Map::new())); + assert_eq!(persisted[0].event_type, "ip.commit.requested"); + assert_eq!(persisted[1].details["buyer"], "bystander"); + + let verification = verify_persisted_chain(temp.path(), &key).unwrap(); + assert_eq!(verification.status, AuditIntegrityStatus::Valid); + } + + #[tokio::test] + async fn anonymize_actor_is_noop_for_unknown_actor() { + let key = vec![5; 32]; + let (store, temp) = open_store("anonymize-noop", key.clone()); + store + .append(AuditEventInput::new(AuditEventCategory::Admin, "admin.action")) + .await + .unwrap(); + let before = fs::read_to_string(temp.path()).unwrap(); + + let count = store.anonymize_actor(&store.hash_identifier("nobody")).await.unwrap(); + + assert_eq!(count, 0); + assert_eq!(fs::read_to_string(temp.path()).unwrap(), before); + } } diff --git a/api-server/src/auth_2fa.rs b/api-server/src/auth_2fa.rs index 836665ee..57b132a2 100644 --- a/api-server/src/auth_2fa.rs +++ b/api-server/src/auth_2fa.rs @@ -98,6 +98,12 @@ impl TwoFactorStore { Ok(store.get(user_id).cloned()) } + /// Remove user's 2FA configuration (#1107). Returns whether one existed. + pub fn remove_config(&self, user_id: &str) -> Result { + let mut store = self.configs.lock().map_err(|e| e.to_string())?; + Ok(store.remove(user_id).is_some()) + } + /// Check if user has 2FA enabled pub fn has_2fa_enabled(&self, user_id: &str) -> Result { Ok(self diff --git a/api-server/src/commitments.rs b/api-server/src/commitments.rs index 0afd1bf4..fc550b2f 100644 --- a/api-server/src/commitments.rs +++ b/api-server/src/commitments.rs @@ -24,6 +24,11 @@ pub fn set_tags(ip_id: u64, tags: Vec) -> Result, String> { Ok(normalized) } +/// Drop all tags for `ip_id` (#1107 cascade). Returns whether any existed. +pub fn remove_tags(ip_id: u64) -> bool { + TAGS.remove(&ip_id).is_some() +} + pub fn get_tags(ip_id: u64) -> Vec { TAGS.get(&ip_id).map(|tags| tags.clone()).unwrap_or_default() } diff --git a/api-server/src/handlers.rs b/api-server/src/handlers.rs index 69bbf272..63ad1f8d 100644 --- a/api-server/src/handlers.rs +++ b/api-server/src/handlers.rs @@ -51,12 +51,12 @@ const SWAP_EXPIRY_SECONDS: u64 = 604800; /// Process-local swap ID counter standing in for the contract's `NextId` /// until the handlers are wired to a live Soroban RPC client. static NEXT_SWAP_ID: AtomicU64 = AtomicU64::new(0); -static WATCHLISTS: Lazy>>> = +pub(crate) static WATCHLISTS: Lazy>>> = Lazy::new(|| Mutex::new(HashMap::new())); -static COMMITMENT_INDEX: Lazy> = Lazy::new(DashMap::new); +pub(crate) static COMMITMENT_INDEX: Lazy> = Lazy::new(DashMap::new); /// Audit log store for tracking all API access and sensitive operations -static AUDIT_LOG_STORE: Lazy> = Lazy::new(|| { +pub(crate) static AUDIT_LOG_STORE: Lazy> = Lazy::new(|| { let audit_key = std::env::var("AUDIT_HMAC_KEY") .unwrap_or_else(|_| "default_audit_key_for_testing".to_string()); let audit_path = std::env::var("AUDIT_LOG_PATH") @@ -482,7 +482,7 @@ pub async fn batch_reveal_commitments( output } - let results = body + let results: Vec = body .commitments .into_iter() .map(|item| { @@ -1574,7 +1574,7 @@ pub async fn execute_batch_swaps( // ── #984: Two-Factor Authentication ──────────────────────────────────────── -static TWO_FACTOR_STORE: once_cell::sync::Lazy = +pub(crate) static TWO_FACTOR_STORE: once_cell::sync::Lazy = once_cell::sync::Lazy::new(crate::auth_2fa::TwoFactorStore::new); /// #984: Enable 2FA for a user @@ -1798,7 +1798,7 @@ pub async fn use_backup_code( // ── #985: Session Management ─────────────────────────────────────────────── -static SESSION_STORE: once_cell::sync::Lazy = +pub(crate) static SESSION_STORE: once_cell::sync::Lazy = once_cell::sync::Lazy::new(|| crate::session::SessionStore::new(crate::session::SessionConfig::default())); /// #985: Check current session status diff --git a/api-server/src/main.rs b/api-server/src/main.rs index 053e7b74..1320a3c2 100644 --- a/api-server/src/main.rs +++ b/api-server/src/main.rs @@ -76,6 +76,7 @@ mod request_queue; mod rate_limit; mod validation; mod validation_middleware; +mod user_data; #[cfg(test)] mod validation_fuzz_tests; #[cfg(test)] @@ -293,6 +294,7 @@ async fn main() { .route("/swap/{swap_id}/reveal", post(handlers::reveal_key)) .route("/swap/{swap_id}/cancel", post(handlers::cancel_swap)) .route("/swap/{swap_id}/cancel-expired", post(handlers::cancel_expired_swap)) + .route("/users/{user_id}", axum::routing::delete(user_data::delete_user_handler)) .route_layer(middleware::from_fn(request_signing::verify_request_signature)); let app = Router::new() diff --git a/api-server/src/session.rs b/api-server/src/session.rs index 5118095e..594df70f 100644 --- a/api-server/src/session.rs +++ b/api-server/src/session.rs @@ -214,6 +214,20 @@ impl SessionStore { Ok(()) } + /// Remove every session belonging to `user_id` (#1107). Returns the count removed. + pub fn remove_user_sessions(&self, user_id: &str) -> Result { + let mut store = self.sessions.lock().map_err(|e| e.to_string())?; + let before = store.len(); + store.retain(|_, session| session.user_id != user_id); + Ok(before - store.len()) + } + + /// Count sessions still held for `user_id`. + pub fn count_user_sessions(&self, user_id: &str) -> Result { + let store = self.sessions.lock().map_err(|e| e.to_string())?; + Ok(store.values().filter(|session| session.user_id == user_id).count()) + } + /// Clean up expired sessions pub fn cleanup_expired_sessions(&self) -> Result { let mut store = self.sessions.lock().map_err(|e| e.to_string())?; diff --git a/api-server/src/user_data.rs b/api-server/src/user_data.rs new file mode 100644 index 00000000..d6392548 --- /dev/null +++ b/api-server/src/user_data.rs @@ -0,0 +1,301 @@ +//! GDPR-style erasure of a user's off-chain data (#1107). +//! +//! `DELETE /users/{user_id}` removes everything this API holds about a user: +//! the commitment index entries and tags for IPs they own (cascading to every +//! watchlist that referenced them), their own watchlist, their 2FA +//! configuration and sessions, and personal data in the audit log. +//! +//! Commitments recorded on-chain are immutable and cannot be deleted by the +//! API. The report says so explicitly via `onchain_commitments_retained`. + +use axum::{ + extract::Path, + http::StatusCode, + Json, +}; +use serde::Serialize; +use std::collections::BTreeSet; +use tracing::instrument; +use utoipa::ToSchema; + +use crate::audit::{AuditEventCategory, AuditEventInput}; +use crate::cache; +use crate::commitments; +use crate::handlers::{AUDIT_LOG_STORE, COMMITMENT_INDEX, SESSION_STORE, TWO_FACTOR_STORE, WATCHLISTS}; +use crate::schemas::ErrorResponse; + +/// Event type recorded in the audit log once an erasure completes. It carries +/// counts only, never the erased user's identifier. +const ERASURE_EVENT_TYPE: &str = "user.erased"; + +#[derive(Debug, Clone, Serialize, ToSchema, PartialEq, Eq)] +pub struct UserErasureReport { + pub user_id: String, + /// Commitments owned by the user that were removed from the off-chain index. + pub commitments_removed: usize, + /// Watchlist entries removed: the user's own list plus references to their IPs. + pub watchlist_entries_removed: usize, + pub tags_removed: usize, + /// Cache entries invalidated for the user's IP records. + pub cache_entries_invalidated: usize, + pub two_factor_configs_removed: usize, + pub sessions_removed: usize, + /// Audit events whose personal data was redacted. + pub audit_events_anonymized: usize, + /// Number of references still found by the post-erasure scan. Zero means + /// the erasure is complete. + pub remaining_references: usize, + pub verified: bool, + /// Always `true`: on-chain commitments cannot be deleted through the API. + pub onchain_commitments_retained: bool, +} + +/// Removes all off-chain data for `user_id` and verifies that nothing remains. +pub async fn erase_user_data(user_id: &str) -> Result { + // Collect owned IPs first so the cascade has a fixed set to work from. + let owned_ids: BTreeSet = COMMITMENT_INDEX + .iter() + .filter(|entry| entry.value().owner == user_id) + .map(|entry| *entry.key()) + .collect(); + + for ip_id in &owned_ids { + COMMITMENT_INDEX.remove(ip_id); + } + + let mut tags_removed = 0; + let mut cache_entries_invalidated = 0; + for ip_id in &owned_ids { + if commitments::remove_tags(*ip_id) { + tags_removed += 1; + } + cache::invalidate(&cache::ip_key(*ip_id)); + cache_entries_invalidated += 1; + } + + let watchlist_entries_removed = { + let mut watchlists = WATCHLISTS + .lock() + .map_err(|_| "watchlist store unavailable".to_string())?; + let mut removed = 0; + if let Some(own) = watchlists.remove(user_id) { + removed += own.len(); + } + for ids in watchlists.values_mut() { + let before = ids.len(); + ids.retain(|ip_id| !owned_ids.contains(ip_id)); + removed += before - ids.len(); + } + removed + }; + + let two_factor_configs_removed = usize::from( + TWO_FACTOR_STORE + .remove_config(user_id) + .map_err(|e| format!("2FA store unavailable: {e}"))?, + ); + + let sessions_removed = SESSION_STORE + .remove_user_sessions(user_id) + .map_err(|e| format!("session store unavailable: {e}"))?; + + let actor_hash = AUDIT_LOG_STORE.hash_identifier(user_id); + let audit_events_anonymized = AUDIT_LOG_STORE + .anonymize_actor(&actor_hash) + .await + .map_err(|e| format!("audit log anonymization failed: {e}"))?; + + let remaining_references = count_remaining_references(user_id, &actor_hash, &owned_ids).await?; + + let report = UserErasureReport { + user_id: user_id.to_string(), + commitments_removed: owned_ids.len(), + watchlist_entries_removed, + tags_removed, + cache_entries_invalidated, + two_factor_configs_removed, + sessions_removed, + audit_events_anonymized, + remaining_references, + verified: remaining_references == 0, + onchain_commitments_retained: true, + }; + + AUDIT_LOG_STORE + .append( + AuditEventInput::new(AuditEventCategory::Admin, ERASURE_EVENT_TYPE) + .detail("commitments_removed", report.commitments_removed) + .detail("audit_events_anonymized", report.audit_events_anonymized) + .detail("verified", report.verified), + ) + .await + .map_err(|e| format!("failed to record erasure in audit log: {e}"))?; + + Ok(report) +} + +/// Re-scans every store for anything still attributable to `user_id`. +async fn count_remaining_references( + user_id: &str, + actor_hash: &str, + owned_ids: &BTreeSet, +) -> Result { + let mut remaining = 0; + + remaining += COMMITMENT_INDEX + .iter() + .filter(|entry| entry.value().owner == user_id) + .count(); + + remaining += owned_ids + .iter() + .filter(|ip_id| !commitments::get_tags(**ip_id).is_empty()) + .count(); + + { + let watchlists = WATCHLISTS + .lock() + .map_err(|_| "watchlist store unavailable".to_string())?; + if watchlists.contains_key(user_id) { + remaining += 1; + } + remaining += watchlists + .values() + .map(|ids| ids.iter().filter(|ip_id| owned_ids.contains(*ip_id)).count()) + .sum::(); + } + + remaining += usize::from( + TWO_FACTOR_STORE + .get_config(user_id) + .map_err(|e| format!("2FA store unavailable: {e}"))? + .is_some(), + ); + remaining += SESSION_STORE + .count_user_sessions(user_id) + .map_err(|e| format!("session store unavailable: {e}"))?; + remaining += AUDIT_LOG_STORE.count_actor_events(actor_hash).await; + + Ok(remaining) +} + +/// Erase all off-chain data held for a user. +#[utoipa::path( + delete, + path = "/users/{user_id}", + tag = "Users", + params(("user_id" = String, Path, description = "User identifier to erase")), + responses( + (status = 200, description = "User data erased", body = UserErasureReport), + (status = 400, description = "Invalid user ID", body = ErrorResponse), + (status = 500, description = "Erasure failed", body = ErrorResponse), + ) +)] +#[instrument] +pub async fn delete_user_handler( + Path(user_id): Path, +) -> Result, (StatusCode, Json)> { + let user_id = user_id.trim(); + if user_id.is_empty() { + return Err(( + StatusCode::BAD_REQUEST, + Json(ErrorResponse { error: "user_id must not be empty".to_string() }), + )); + } + + let report = erase_user_data(user_id).await.map_err(|error| { + tracing::error!(error = %error, "user data erasure failed"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ErrorResponse { error: "user data erasure failed".to_string() }), + ) + })?; + + if !report.verified { + tracing::error!(remaining = report.remaining_references, "user data erasure incomplete"); + } + + Ok(Json(report)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::schemas::IpRecord; + + fn unique(label: &str) -> String { + format!("erase-test-{label}-{}", uuid::Uuid::new_v4()) + } + + fn next_ip_id() -> u64 { + uuid::Uuid::new_v4().as_u128() as u64 + } + + fn register_commitment_for_tests(owner: &str) -> u64 { + let ip_id = next_ip_id(); + COMMITMENT_INDEX.insert( + ip_id, + IpRecord { + ip_id, + owner: owner.to_string(), + commitment_hash: "ab".repeat(32), + timestamp: 0, + revoked: false, + }, + ); + ip_id + } + + fn add_to_watchlist_for_tests(user_id: &str, ip_id: u64) { + WATCHLISTS + .lock() + .unwrap() + .entry(user_id.to_string()) + .or_default() + .insert(ip_id); + } + + #[tokio::test] + async fn erasure_cascades_to_owned_commitments_and_watchlists() { + let owner = unique("owner"); + let viewer = unique("viewer"); + let owned_ip = register_commitment_for_tests(&owner); + let other_ip = register_commitment_for_tests(&unique("other-owner")); + add_to_watchlist_for_tests(&viewer, owned_ip); + add_to_watchlist_for_tests(&viewer, other_ip); + add_to_watchlist_for_tests(&owner, other_ip); + commitments::set_tags(owned_ip, vec!["patent".into()]).unwrap(); + + let report = erase_user_data(&owner).await.unwrap(); + + assert_eq!(report.commitments_removed, 1); + assert_eq!(report.tags_removed, 1); + assert!(report.watchlist_entries_removed >= 2); + assert!(report.verified, "remaining references: {}", report.remaining_references); + assert!(report.onchain_commitments_retained); + + assert!(!COMMITMENT_INDEX.contains_key(&owned_ip)); + assert!(commitments::get_tags(owned_ip).is_empty()); + let watchlists = WATCHLISTS.lock().unwrap(); + assert!(!watchlists.get(&viewer).map_or(false, |ids| ids.contains(&owned_ip))); + assert!(watchlists.get(&viewer).map_or(false, |ids| ids.contains(&other_ip))); + assert!(!watchlists.contains_key(&owner)); + } + + #[tokio::test] + async fn erasure_is_idempotent_for_unknown_user() { + let report = erase_user_data(&unique("unknown")).await.unwrap(); + + assert_eq!(report.commitments_removed, 0); + assert_eq!(report.remaining_references, 0); + assert!(report.verified); + } + + #[tokio::test] + async fn handler_rejects_blank_user_id() { + let result = delete_user_handler(Path(" ".to_string())).await; + + let (status, _) = result.unwrap_err(); + assert_eq!(status, StatusCode::BAD_REQUEST); + } +} From db3b9d60642ea0d87e65cf933e3d113f83f200b1 Mon Sep 17 00:00:00 2001 From: usman222nuhu Date: Fri, 16 Oct 2026 05:50:18 +0000 Subject: [PATCH 2/2] feat(api-server): add threat detection for suspicious activity (#1108) Adds sliding-window detectors that alert once per burst when: - one owner registers 5 commitments within 60s (rapid-fire commits) - one actor touches 50 distinct IP records within 60s (unusual access) - the same IP accumulates 5 failed verifications within 60s (suspicious verification chain) Alerts are logged and submitted to the alert manager. Commit and batch reveal handlers feed the detectors. --- api-server/src/handlers.rs | 23 +++ api-server/src/main.rs | 1 + api-server/src/threat_detection.rs | 308 +++++++++++++++++++++++++++++ 3 files changed, 332 insertions(+) create mode 100644 api-server/src/threat_detection.rs diff --git a/api-server/src/handlers.rs b/api-server/src/handlers.rs index 63ad1f8d..ae35d771 100644 --- a/api-server/src/handlers.rs +++ b/api-server/src/handlers.rs @@ -118,6 +118,13 @@ pub async fn commit_ip( ) })?; + // #1108: feed the successful commit into rapid-fire detection. + if let Some(alert) = crate::threat_detection::THREAT_DETECTOR + .record_commit(&body.owner, crate::alerting::now_secs()) + { + crate::threat_detection::dispatch(&alert); + } + // #983: Check per-user commitment rate limit after successful commit let (allowed, _remaining, reset_after) = rate_limiter.check_commitment_rate_limit(&body.owner).await; if !allowed { @@ -237,6 +244,12 @@ pub async fn transfer_ip(Json(body): Json) -> Result) -> Result, (StatusCode, Json)> { // TODO: Call Soroban RPC to invoke ip_registry.verify_commitment + // #1108: this path currently always fails, so each attempt counts as a failed verification. + if let Some(alert) = crate::threat_detection::THREAT_DETECTOR + .record_verification(body.ip_id, false, crate::alerting::now_secs()) + { + crate::threat_detection::dispatch(&alert); + } Err(( StatusCode::NOT_FOUND, Json(ErrorResponse { @@ -516,6 +529,16 @@ pub async fn batch_reveal_commitments( }) .collect(); + // #1108: each reveal outcome feeds the verification-chain detector. + let now = crate::alerting::now_secs(); + for result in &results { + if let Some(alert) = crate::threat_detection::THREAT_DETECTOR + .record_verification(result.ip_id, result.valid, now) + { + crate::threat_detection::dispatch(&alert); + } + } + Ok(Json(BatchRevealCommitmentsResponse { results })) } diff --git a/api-server/src/main.rs b/api-server/src/main.rs index 1320a3c2..e15f4a27 100644 --- a/api-server/src/main.rs +++ b/api-server/src/main.rs @@ -77,6 +77,7 @@ mod rate_limit; mod validation; mod validation_middleware; mod user_data; +mod threat_detection; #[cfg(test)] mod validation_fuzz_tests; #[cfg(test)] diff --git a/api-server/src/threat_detection.rs b/api-server/src/threat_detection.rs new file mode 100644 index 00000000..2c2bf41e --- /dev/null +++ b/api-server/src/threat_detection.rs @@ -0,0 +1,308 @@ +//! Threat detection for suspicious activity (#1108). +//! +//! Three sliding-window detectors run over request activity: +//! +//! * **Rapid-fire commitments** — one owner registers `rapid_commit_threshold` +//! commitments inside `window_secs`. +//! * **Unusual access** — one actor touches `unusual_access_threshold` distinct +//! IP records inside `window_secs` (enumeration / scraping). +//! * **Suspicious verification chains** — `failed_verification_threshold` +//! failed verifications against the same IP inside `window_secs` +//! (guessing a commitment's secret or blinding factor). +//! +//! Each detector fires once when its threshold is crossed, then re-arms after +//! the window drains, so a sustained attack produces one alert per burst +//! rather than one per request. Time is passed in explicitly so the logic is +//! deterministic under test. + +use crate::alerting::{Alert, Severity, ALERT_MANAGER}; +use once_cell::sync::Lazy; +use serde::Serialize; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::Mutex; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ThreatKind { + RapidCommitments, + UnusualAccess, + SuspiciousVerification, +} + +impl ThreatKind { + pub fn as_str(&self) -> &'static str { + match self { + Self::RapidCommitments => "rapid_commitments", + Self::UnusualAccess => "unusual_access", + Self::SuspiciousVerification => "suspicious_verification", + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ThreatAlert { + pub kind: ThreatKind, + /// Owner address for commit/access threats, `ip:` for verification threats. + pub subject: String, + pub count: usize, + pub window_secs: u64, + pub detected_at: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ThreatConfig { + pub window_secs: u64, + pub rapid_commit_threshold: usize, + pub unusual_access_threshold: usize, + pub failed_verification_threshold: usize, +} + +impl Default for ThreatConfig { + fn default() -> Self { + Self { + window_secs: 60, + rapid_commit_threshold: 5, + unusual_access_threshold: 50, + failed_verification_threshold: 5, + } + } +} + +/// Process-wide detector used by the request handlers. +pub static THREAT_DETECTOR: Lazy = + Lazy::new(|| ThreatDetector::new(ThreatConfig::default())); + +pub struct ThreatDetector { + config: ThreatConfig, + commits: Mutex>>, + accesses: Mutex>>, + failed_verifications: Mutex>>, +} + +/// Drops timestamps that have aged out of the window ending at `now`. +fn prune_timestamps(entries: &mut VecDeque, now: u64, window_secs: u64) { + let cutoff = now.saturating_sub(window_secs); + while entries.front().is_some_and(|&ts| ts < cutoff) { + entries.pop_front(); + } +} + +impl ThreatDetector { + pub fn new(config: ThreatConfig) -> Self { + Self { + config, + commits: Mutex::new(HashMap::new()), + accesses: Mutex::new(HashMap::new()), + failed_verifications: Mutex::new(HashMap::new()), + } + } + + /// Records a commitment registration by `owner`. + pub fn record_commit(&self, owner: &str, now: u64) -> Option { + let mut commits = self.commits.lock().ok()?; + let entries = commits.entry(owner.to_string()).or_default(); + prune_timestamps(entries, now, self.config.window_secs); + entries.push_back(now); + + if entries.len() == self.config.rapid_commit_threshold { + return Some(ThreatAlert { + kind: ThreatKind::RapidCommitments, + subject: owner.to_string(), + count: entries.len(), + window_secs: self.config.window_secs, + detected_at: now, + }); + } + None + } + + /// Records that `actor` accessed IP record `ip_id`. + pub fn record_access(&self, actor: &str, ip_id: u64, now: u64) -> Option { + let mut accesses = self.accesses.lock().ok()?; + let entries = accesses.entry(actor.to_string()).or_default(); + let cutoff = now.saturating_sub(self.config.window_secs); + while entries.front().is_some_and(|&(ts, _)| ts < cutoff) { + entries.pop_front(); + } + entries.push_back((now, ip_id)); + + let distinct: HashSet = entries.iter().map(|&(_, id)| id).collect(); + if distinct.len() == self.config.unusual_access_threshold { + return Some(ThreatAlert { + kind: ThreatKind::UnusualAccess, + subject: actor.to_string(), + count: distinct.len(), + window_secs: self.config.window_secs, + detected_at: now, + }); + } + None + } + + /// Records the outcome of verifying IP record `ip_id`. Successful + /// verifications are not tracked; only failures count toward the chain. + pub fn record_verification(&self, ip_id: u64, valid: bool, now: u64) -> Option { + if valid { + return None; + } + + let mut failures = self.failed_verifications.lock().ok()?; + let entries = failures.entry(ip_id).or_default(); + prune_timestamps(entries, now, self.config.window_secs); + entries.push_back(now); + + if entries.len() == self.config.failed_verification_threshold { + return Some(ThreatAlert { + kind: ThreatKind::SuspiciousVerification, + subject: format!("ip:{ip_id}"), + count: entries.len(), + window_secs: self.config.window_secs, + detected_at: now, + }); + } + None + } +} + +/// Reports a detected threat to logs and the alert manager. +pub fn dispatch(alert: &ThreatAlert) { + tracing::warn!( + kind = alert.kind.as_str(), + subject = %alert.subject, + count = alert.count, + window_secs = alert.window_secs, + "threat detected" + ); + + let summary = format!( + "{} for {}: {} events within {}s", + alert.kind.as_str(), + alert.subject, + alert.count, + alert.window_secs + ); + let _ = ALERT_MANAGER.submit( + Alert::new( + format!("threat_{}", alert.kind.as_str()), + Severity::Critical, + summary, + ) + .with_label("subject", alert.subject.clone()), + ); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn detector() -> ThreatDetector { + ThreatDetector::new(ThreatConfig { + window_secs: 60, + rapid_commit_threshold: 3, + unusual_access_threshold: 4, + failed_verification_threshold: 3, + }) + } + + #[test] + fn rapid_commits_alert_when_threshold_reached_within_window() { + let d = detector(); + assert!(d.record_commit("GOWNER", 100).is_none()); + assert!(d.record_commit("GOWNER", 101).is_none()); + + let alert = d.record_commit("GOWNER", 102).expect("third commit in window alerts"); + assert_eq!(alert.kind, ThreatKind::RapidCommitments); + assert_eq!(alert.subject, "GOWNER"); + assert_eq!(alert.count, 3); + } + + #[test] + fn spaced_commits_do_not_alert() { + let d = detector(); + assert!(d.record_commit("GOWNER", 0).is_none()); + assert!(d.record_commit("GOWNER", 100).is_none()); + assert!(d.record_commit("GOWNER", 200).is_none()); + } + + #[test] + fn commit_alert_fires_once_per_burst_then_rearms_after_window() { + let d = detector(); + d.record_commit("A", 0); + d.record_commit("A", 1); + assert!(d.record_commit("A", 2).is_some()); + assert!(d.record_commit("A", 3).is_none(), "no alert per extra request"); + + // Old events age out, so a fresh burst alerts again. + assert!(d.record_commit("A", 500).is_none()); + assert!(d.record_commit("A", 501).is_none()); + assert!(d.record_commit("A", 502).is_some()); + } + + #[test] + fn commit_counts_are_tracked_per_owner() { + let d = detector(); + d.record_commit("A", 0); + d.record_commit("B", 0); + assert!(d.record_commit("A", 1).is_none()); + assert!(d.record_commit("B", 1).is_none()); + } + + #[test] + fn unusual_access_counts_distinct_ips_not_repeat_requests() { + let d = detector(); + for _ in 0..10 { + assert!(d.record_access("scraper", 7, 0).is_none()); + } + d.record_access("scraper", 8, 1); + d.record_access("scraper", 9, 2); + let alert = d.record_access("scraper", 10, 3).expect("fourth distinct IP alerts"); + assert_eq!(alert.kind, ThreatKind::UnusualAccess); + assert_eq!(alert.count, 4); + } + + #[test] + fn unusual_access_forgets_ips_outside_the_window() { + let d = detector(); + d.record_access("scraper", 1, 0); + d.record_access("scraper", 2, 0); + d.record_access("scraper", 3, 0); + // Only the window-recent accesses should count. + assert!(d.record_access("scraper", 4, 1000).is_none()); + } + + #[test] + fn failed_verifications_alert_on_repeated_failures_for_same_ip() { + let d = detector(); + assert!(d.record_verification(42, false, 0).is_none()); + assert!(d.record_verification(42, false, 1).is_none()); + let alert = d.record_verification(42, false, 2).expect("third failure alerts"); + assert_eq!(alert.kind, ThreatKind::SuspiciousVerification); + assert_eq!(alert.subject, "ip:42"); + } + + #[test] + fn successful_verifications_never_count_as_failures() { + let d = detector(); + for t in 0..20 { + assert!(d.record_verification(42, true, t).is_none()); + } + } + + #[test] + fn failures_are_isolated_per_ip() { + let d = detector(); + d.record_verification(1, false, 0); + d.record_verification(2, false, 0); + assert!(d.record_verification(1, false, 1).is_none()); + assert!(d.record_verification(2, false, 1).is_none()); + } + + #[test] + fn default_config_matches_documented_thresholds() { + let config = ThreatConfig::default(); + assert_eq!(config.window_secs, 60); + assert_eq!(config.rapid_commit_threshold, 5); + assert_eq!(config.unusual_access_threshold, 50); + assert_eq!(config.failed_verification_threshold, 5); + } +}