diff --git a/net-guardia/src/adapter/http/flow_trace.rs b/net-guardia/src/adapter/http/flow_trace.rs index 656bfb9..c9cf9c9 100644 --- a/net-guardia/src/adapter/http/flow_trace.rs +++ b/net-guardia/src/adapter/http/flow_trace.rs @@ -105,7 +105,6 @@ pub fn is_safe_flow_trace_name(name: &str) -> bool { /// `Engine` if the logger is active. Returns `None` when Flow Trace /// isn't enabled (Dormant state). fn flow_trace_directory(engine: &web::Data) -> Option { - let _ = engine; // placeholder until Engine exposes logger directory engine.traffic_logger_directory().map(Path::to_path_buf) } diff --git a/net-guardia/src/adapter/http/fusion.rs b/net-guardia/src/adapter/http/fusion.rs index 6eb953a..b243f81 100644 --- a/net-guardia/src/adapter/http/fusion.rs +++ b/net-guardia/src/adapter/http/fusion.rs @@ -1,16 +1,38 @@ -//! HTTP surface for fusion-layer observability. Reads shared atomic -//! counters maintained by the detection orchestrator — handlers here -//! never touch orchestrator state, so a hung dashboard cannot stall -//! the detection pipeline. +//! HTTP surface for fusion-layer observability + incident explain. +//! Metrics handlers read shared atomic counters maintained by the +//! detection orchestrator — they never touch orchestrator state, so a +//! hung dashboard cannot stall the detection pipeline. The explain +//! handler reads the WORM audit chain populated by +//! `publish_fusion_audit` and surfaces a per-IP evidence timeline so +//! analysts can answer "why was this IP blocked?" without parsing +//! logs by hand. use std::sync::Arc; -use actix_web::{HttpResponse, Responder, Scope, web}; +use actix_web::{HttpRequest, HttpResponse, Responder, Scope, web}; use crate::core::detection::metrics::FusionMetrics; +use crate::interface::port::audit::AuditRepo; + +/// Maximum audit rows scanned per explain request. Caps DB work in +/// case the audit chain grows large enough that a naive full-table +/// scan would be noticeable. +const FUSION_EXPLAIN_SCAN_LIMIT: i64 = 5_000; + +/// Upper cap on entries returned to the client per explain request. +/// Guards against a UI rendering path that chokes on enormous JSON. +const FUSION_EXPLAIN_RESPONSE_CAP: usize = 200; + +/// Stable audit action string the fusion engine emits — kept in sync +/// with `core::detection::orchestrator::FUSION_AUDIT_ACTION`. If that +/// constant changes, the explain endpoint silently returns nothing, so +/// keep this updated at the same time. +const FUSION_AUDIT_ACTION: &str = "fused_threat_emitted"; pub fn initialize() -> Scope { - web::scope("/fusion").route("/metrics", web::get().to(get_metrics)) + web::scope("/fusion") + .route("/metrics", web::get().to(get_metrics)) + .route("/explain/{src_ip}", web::get().to(explain_ip)) } /// `GET /api/fusion/metrics` — lock-free snapshot of fusion counters and @@ -19,3 +41,166 @@ pub fn initialize() -> Scope { async fn get_metrics(metrics: web::Data>) -> impl Responder { HttpResponse::Ok().json(metrics.snapshot()) } + +/// `GET /api/fusion/explain/{src_ip}` — per-IP fusion evidence timeline. +/// Scans the WORM audit chain for `fused_threat_emitted` entries that +/// match `src_ip`, returning them oldest-first so the UI can render a +/// chronological "why was this IP blocked" view. +async fn explain_ip(req: HttpRequest, audit: web::Data>) -> impl Responder { + let src_ip = match req.match_info().get("src_ip") { + Some(ip) => ip.to_string(), + None => { + return HttpResponse::BadRequest().json(serde_json::json!({ + "error": "missing src_ip path segment", + })); + } + }; + + let entries = match audit.list_audit_logs_by_action(FUSION_AUDIT_ACTION, FUSION_EXPLAIN_SCAN_LIMIT) { + Ok(e) => e, + Err(e) => { + return HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("audit store unavailable: {e}"), + })); + } + }; + + let matches = filter_fusion_evidence_for_ip(&entries, &src_ip, FUSION_EXPLAIN_RESPONSE_CAP); + HttpResponse::Ok().json(serde_json::json!({ + "src_ip": src_ip, + "match_count": matches.len(), + "truncated": matches.len() >= FUSION_EXPLAIN_RESPONSE_CAP, + "entries": matches, + })) +} + +/// Filter audit entries down to the ones whose JSON detail's `src_ip` +/// matches `target_ip`, ordered oldest-first (ascending id). Entries +/// with unparseable detail are dropped silently — the chain is +/// append-only, so a malformed row is an integrity concern for the +/// audit-verify endpoint to surface, not this handler. +/// +/// Extracted as a free function so tests can cover the filter / +/// ordering / cap behaviour without an in-memory DB. +pub fn filter_fusion_evidence_for_ip( + entries: &[crate::interface::port::audit::AuditLogEntry], + target_ip: &str, + cap: usize, +) -> Vec { + let mut filtered: Vec<&crate::interface::port::audit::AuditLogEntry> = entries + .iter() + .filter(|entry| detail_matches_src_ip(&entry.detail, target_ip)) + .collect(); + filtered.sort_by_key(|entry| entry.id); + filtered + .into_iter() + .take(cap) + .map(|entry| { + let detail: serde_json::Value = serde_json::from_str(&entry.detail).unwrap_or(serde_json::Value::Null); + serde_json::json!({ + "id": entry.id, + "actor": entry.actor, + "action": entry.action, + "created_at": entry.created_at, + "detail": detail, + }) + }) + .collect() +} + +fn detail_matches_src_ip(detail_json: &str, target_ip: &str) -> bool { + let parsed: serde_json::Value = match serde_json::from_str(detail_json) { + Ok(v) => v, + Err(_) => return false, + }; + parsed.get("src_ip").and_then(|v| v.as_str()) == Some(target_ip) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::interface::port::audit::AuditLogEntry; + + fn entry(id: i64, src_ip: &str, attack: &str) -> AuditLogEntry { + let detail = serde_json::json!({ + "src_ip": src_ip, + "attack_type": attack, + "fused_confidence": 0.9, + "per_source": [{"source": "Suricata", "confidence": 0.9, "local_attack_type": "brute-force"}], + }) + .to_string(); + AuditLogEntry { + id, + actor: "FusionEngine".to_string(), + action: "fused_threat_emitted".to_string(), + detail, + created_at: format!("2026-04-18T10:00:{:02}Z", id), + } + } + + #[test] + fn filter_returns_only_matching_src_ip() { + let entries = [ + entry(1, "1.2.3.4", "brute_force"), + entry(2, "10.0.0.5", "port_scan"), + entry(3, "1.2.3.4", "exploit"), + ]; + let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100); + assert_eq!(got.len(), 2); + assert_eq!(got[0]["id"], 1); + assert_eq!(got[1]["id"], 3); + } + + #[test] + fn filter_sorts_oldest_first_even_when_input_is_reversed() { + // Real repo query returns DESC; filter must still hand back ASC. + let entries = [ + entry(30, "1.1.1.1", "a"), + entry(10, "1.1.1.1", "b"), + entry(20, "1.1.1.1", "c"), + ]; + let got = filter_fusion_evidence_for_ip(&entries, "1.1.1.1", 100); + let ids: Vec = got.iter().map(|v| v["id"].as_i64().unwrap()).collect(); + assert_eq!(ids, vec![10, 20, 30]); + } + + #[test] + fn filter_applies_response_cap() { + let entries: Vec = (1..=10).map(|i| entry(i, "9.9.9.9", "x")).collect(); + let got = filter_fusion_evidence_for_ip(&entries, "9.9.9.9", 3); + assert_eq!(got.len(), 3); + let ids: Vec = got.iter().map(|v| v["id"].as_i64().unwrap()).collect(); + assert_eq!(ids, vec![1, 2, 3], "cap takes oldest, not newest"); + } + + #[test] + fn filter_drops_rows_with_unparseable_detail() { + let good = entry(1, "1.2.3.4", "brute_force"); + let bad = AuditLogEntry { + id: 2, + actor: "FusionEngine".into(), + action: "fused_threat_emitted".into(), + detail: "{{not json".into(), + created_at: "2026-04-18T10:00:02Z".into(), + }; + let got = filter_fusion_evidence_for_ip(&[good, bad], "1.2.3.4", 100); + assert_eq!(got.len(), 1); + assert_eq!(got[0]["id"], 1); + } + + #[test] + fn filter_nonmatching_ip_returns_empty() { + let entries = [entry(1, "1.2.3.4", "brute_force")]; + assert!(filter_fusion_evidence_for_ip(&entries, "5.6.7.8", 100).is_empty()); + } + + #[test] + fn filter_preserves_detail_structure_in_response() { + let entries = [entry(1, "1.2.3.4", "brute_force")]; + let got = filter_fusion_evidence_for_ip(&entries, "1.2.3.4", 100); + assert_eq!(got.len(), 1); + let detail = &got[0]["detail"]; + assert_eq!(detail["attack_type"], "brute_force"); + assert_eq!(detail["per_source"][0]["source"], "Suricata"); + } +} diff --git a/net-guardia/src/adapter/persistence/repository.rs b/net-guardia/src/adapter/persistence/repository.rs index 0d7b121..ee8415e 100644 --- a/net-guardia/src/adapter/persistence/repository.rs +++ b/net-guardia/src/adapter/persistence/repository.rs @@ -366,7 +366,11 @@ impl Database { "system:admin", "users:read", "users:write", - "users:admin" + "users:admin", + "fusion:read", + "fusion:write", + "flow_trace:read", + "flow_trace:write" ]) .to_string(); let viewer_permissions = serde_json::json!([ @@ -380,7 +384,9 @@ impl Database { "dns_filter:read", "rate_limit:read", "protocol_filter:read", - "system:read" + "system:read", + "fusion:read", + "flow_trace:read" ]) .to_string(); @@ -1652,6 +1658,28 @@ impl Database { Ok(rows) } + /// Read audit entries whose `action` matches exactly, newest-first, + /// capped at `limit`. Drives the fusion explain endpoint. + pub fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result, Error> { + let conn = self.conn()?; + let mut stmt = conn.prepare( + "SELECT id, actor, action, detail, ts FROM audit_log WHERE action = ?1 ORDER BY id DESC LIMIT ?2", + )?; + let rows = stmt + .query_map(params![action, limit], |row| { + Ok(AuditLogEntry { + id: row.get(0)?, + actor: row.get(1)?, + action: row.get(2)?, + detail: row.get(3)?, + created_at: row.get(4)?, + }) + })? + .filter_map(|r| r.ok()) + .collect(); + Ok(rows) + } + /// Walk the entire audit_log in id order and verify the hash chain. /// Returns `Ok(count)` on success; returns `Err` at the first mismatch, /// naming the offending row id and the kind of mismatch. @@ -2067,6 +2095,9 @@ impl AuditRepo for Database { fn list_audit_logs(&self) -> Result, Error> { self.list_audit_logs() } + fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result, Error> { + self.list_audit_logs_by_action(action, limit) + } fn verify_audit_log_chain(&self) -> Result { self.verify_audit_log_chain() } diff --git a/net-guardia/src/core/auth/middleware.rs b/net-guardia/src/core/auth/middleware.rs index e0854b9..ad8b90d 100644 --- a/net-guardia/src/core/auth/middleware.rs +++ b/net-guardia/src/core/auth/middleware.rs @@ -49,6 +49,10 @@ fn required_permission(path: &str, method: &Method) -> Option { "dashboard" } else if path.starts_with("/api/ml/") { "ai_detection" + } else if path.starts_with("/api/fusion/") { + "fusion" + } else if path.starts_with("/api/flow-trace/") { + "flow_trace" } else if path.starts_with("/api/acl/geo/") { "geo_block" } else if path.starts_with("/api/acl/") { diff --git a/net-guardia/src/infrastructure/http_server.rs b/net-guardia/src/infrastructure/http_server.rs index a66abb8..e42a409 100644 --- a/net-guardia/src/infrastructure/http_server.rs +++ b/net-guardia/src/infrastructure/http_server.rs @@ -35,6 +35,7 @@ use crate::infrastructure::suricata_manager::SuricataManager; use crate::infrastructure::system::ShutdownHandle; use crate::interface::port::api_key::ApiKeyRepo; use crate::interface::port::app_repo::AppRepo; +use crate::interface::port::audit::AuditRepo; use crate::model::config::constants::HTTP_FALLBACK_PORT; use crate::model::error::Error; use crate::model::error::http::HttpError; @@ -155,6 +156,7 @@ pub fn start_setup_server( .wrap(cors(vec![])) .app_data(web::Data::from(db.clone() as Arc)) .app_data(web::Data::from(db.clone() as Arc)) + .app_data(web::Data::from(db.clone() as Arc)) .app_data(web::Data::from(db.clone())) .app_data(web::Data::from(secret_store.clone())) .app_data(web::Data::from(jwt_service.clone())) @@ -255,6 +257,7 @@ pub async fn run(params: HttpServerParams) -> Result<(), Error> { .app_data(web::Data::from(drop_monitor.clone())) .app_data(web::Data::from(db.clone() as Arc)) .app_data(web::Data::from(db.clone() as Arc)) + .app_data(web::Data::from(db.clone() as Arc)) .app_data(web::Data::from(db.clone())) .app_data(web::Data::from(secret_store.clone())) .app_data(web::Data::from(jwt_service.clone())) diff --git a/net-guardia/src/interface/port/audit.rs b/net-guardia/src/interface/port/audit.rs index 350c15a..3292ee4 100644 --- a/net-guardia/src/interface/port/audit.rs +++ b/net-guardia/src/interface/port/audit.rs @@ -23,6 +23,12 @@ pub trait AuditRepo: Send + Sync { /// Read all audit entries ordered by id ASC. fn list_audit_logs(&self) -> Result, Error>; + /// Read audit entries whose `action` exactly matches, newest first, + /// capped at `limit`. Drives the fusion explain endpoint, which + /// filters on `fused_threat_emitted` rather than walking the full + /// chain for every request. + fn list_audit_logs_by_action(&self, action: &str, limit: i64) -> Result, Error>; + /// Walk the full chain and verify every `row_hash` matches /// `H(ts || actor || action || detail || prev_hash)`. Returns the number /// of entries verified. Errors on the first broken link.