From d2d52e84a0cc07ae6d8b761626469058d0e2b621 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 14:49:51 -0400 Subject: [PATCH 1/6] feat(rtbh): shadow executor + no-op journal (record mitigations, don't act) --- crates/blackwall-rtbh/src/lib.rs | 2 + crates/blackwall-rtbh/src/shadow.rs | 238 ++++++++++++++++++++++++++++ 2 files changed, 240 insertions(+) create mode 100644 crates/blackwall-rtbh/src/shadow.rs diff --git a/crates/blackwall-rtbh/src/lib.rs b/crates/blackwall-rtbh/src/lib.rs index 3068bcb..b30a3d6 100644 --- a/crates/blackwall-rtbh/src/lib.rs +++ b/crates/blackwall-rtbh/src/lib.rs @@ -7,6 +7,7 @@ pub mod controller; pub mod flowspec_controller; pub mod flowspec_manager; pub mod manager; +pub mod shadow; pub use controller::{BlackholeOrigin, RtbhAction, RtbhConfig, RtbhController}; pub use flowspec_controller::{FlowKey, FlowSpecAction, FlowSpecConfig, FlowSpecController}; @@ -14,6 +15,7 @@ pub use flowspec_manager::{FlowSpecJournal, FlowSpecManager}; pub use manager::{ ApplyOutcome, BgpError, BgpExecutor, BlackholeJournal, JournalError, RtbhManager, }; +pub use shadow::{NoOpJournal, ShadowAction, ShadowBgpExecutor, ShadowRecorder}; /// Executes BGP commands against a live session via [`blackwall_bgp::BgpHandle`]. #[async_trait::async_trait] diff --git a/crates/blackwall-rtbh/src/shadow.rs b/crates/blackwall-rtbh/src/shadow.rs new file mode 100644 index 0000000..48627a5 --- /dev/null +++ b/crates/blackwall-rtbh/src/shadow.rs @@ -0,0 +1,238 @@ +//! Shadow-mode executor + journal: record what a mitigation *would* do without +//! executing it. Wired in place of the real `BgpExecutor`/journal when the +//! `shadow` config directive is set. + +use crate::controller::BlackholeOrigin; +use crate::flowspec_manager::FlowSpecJournal; +use crate::manager::{BgpError, BgpExecutor, BlackholeJournal, JournalError}; +use async_trait::async_trait; +use std::net::IpAddr; + +/// A mitigation the daemon would have applied, captured for logging/audit. +#[derive(Debug, Clone)] +pub enum ShadowAction { + /// Would announce a blackhole route. + RtbhAnnounce(blackwall_bgp::Route), + /// Would withdraw a blackhole prefix. + RtbhWithdraw(ipnet::IpNet), + /// Would announce a FlowSpec rule. + FlowSpecAnnounce(blackwall_bgp::FlowSpecRule), + /// Would withdraw a FlowSpec rule. + FlowSpecWithdraw(blackwall_bgp::FlowSpecRule), +} + +/// Sink for shadow actions (the concrete impl in `blackwalld` logs + audits + +/// meters; tests capture into a vec). +#[async_trait] +pub trait ShadowRecorder: Send + Sync { + /// Record one intended mitigation. + async fn record(&self, action: ShadowAction); +} + +/// Blanket impl so an `Arc` can be shared with other +/// owners (e.g. a test assertion) while also being handed to +/// [`ShadowBgpExecutor::new`]. +#[async_trait] +impl ShadowRecorder for std::sync::Arc { + async fn record(&self, action: ShadowAction) { + (**self).record(action).await; + } +} + +/// A `BgpExecutor` that records intended announcements instead of sending them. +/// +/// Holds no reference to a real BGP session or executor at all — there is no +/// path from any of its methods to live BGP traffic. +pub struct ShadowBgpExecutor { + recorder: R, +} + +impl ShadowBgpExecutor { + /// Wrap a recorder. + pub fn new(recorder: R) -> Self { + Self { recorder } + } +} + +#[async_trait] +impl BgpExecutor for ShadowBgpExecutor { + async fn announce(&self, route: blackwall_bgp::Route) -> Result<(), BgpError> { + self.recorder + .record(ShadowAction::RtbhAnnounce(route)) + .await; + Ok(()) + } + async fn withdraw(&self, prefix: ipnet::IpNet) -> Result<(), BgpError> { + self.recorder + .record(ShadowAction::RtbhWithdraw(prefix)) + .await; + Ok(()) + } + async fn announce_flowspec(&self, rule: blackwall_bgp::FlowSpecRule) -> Result<(), BgpError> { + self.recorder + .record(ShadowAction::FlowSpecAnnounce(rule)) + .await; + Ok(()) + } + async fn withdraw_flowspec(&self, rule: blackwall_bgp::FlowSpecRule) -> Result<(), BgpError> { + self.recorder + .record(ShadowAction::FlowSpecWithdraw(rule)) + .await; + Ok(()) + } +} + +/// A journal that persists nothing — used with [`ShadowBgpExecutor`] so the +/// live mirror tables stay empty (nothing was actually announced). +pub struct NoOpJournal; + +#[async_trait] +impl BlackholeJournal for NoOpJournal { + async fn record_announce( + &self, + _target: IpAddr, + _origin: BlackholeOrigin, + _at_ms: u64, + ) -> Result<(), JournalError> { + Ok(()) + } + async fn record_withdraw(&self, _target: IpAddr, _at_ms: u64) -> Result<(), JournalError> { + Ok(()) + } +} + +#[async_trait] +impl FlowSpecJournal for NoOpJournal { + async fn record_announce( + &self, + _rule: blackwall_bgp::FlowSpecRule, + _origin: BlackholeOrigin, + _at_ms: u64, + ) -> Result<(), JournalError> { + Ok(()) + } + async fn record_withdraw( + &self, + _rule: blackwall_bgp::FlowSpecRule, + _at_ms: u64, + ) -> Result<(), JournalError> { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::manager::{BgpError, BgpExecutor}; + use blackwall_bgp::{FlowAction, FlowSpecRule, Origin, Route}; + use std::sync::Mutex; + + #[derive(Default)] + struct CapturingRecorder(Mutex>); + #[async_trait::async_trait] + impl ShadowRecorder for CapturingRecorder { + async fn record(&self, action: ShadowAction) { + self.0.lock().unwrap().push(action); + } + } + + /// Never constructed: its whole point is to prove `ShadowBgpExecutor` + /// holds no path to a real `BgpExecutor` for the tests below to wire in. + #[expect( + dead_code, + reason = "documents the absent BGP path; never instantiated on purpose" + )] + struct PanicExecutor; + #[async_trait::async_trait] + impl BgpExecutor for PanicExecutor { + async fn announce(&self, _r: Route) -> Result<(), BgpError> { + panic!("announced in shadow!") + } + async fn withdraw(&self, _p: ipnet::IpNet) -> Result<(), BgpError> { + panic!("withdrew in shadow!") + } + async fn announce_flowspec(&self, _r: FlowSpecRule) -> Result<(), BgpError> { + panic!("announced flowspec in shadow!") + } + async fn withdraw_flowspec(&self, _r: FlowSpecRule) -> Result<(), BgpError> { + panic!("withdrew flowspec in shadow!") + } + } + + /// Build a minimal /32 blackhole route the same way + /// `RtbhController::build_route` does (see `controller.rs`). + fn route(target: &str) -> Route { + Route { + prefix: format!("{target}/32").parse().unwrap(), + next_hop: "10.0.0.1".parse().unwrap(), + origin: Origin::Igp, + communities: vec![(65535, 666)], + large_communities: Vec::new(), + } + } + + /// Build a `FlowSpecRule` the same way `flowspec_controller.rs`'s tests do. + fn flowspec_rule() -> FlowSpecRule { + FlowSpecRule { + dst: "203.0.113.7/32".parse().unwrap(), + protocol: Some(17), + dst_port: Some(53), + action: FlowAction::TrafficRate(1000.0), + } + } + + #[tokio::test] + async fn announce_records_and_never_calls_bgp() { + let rec = std::sync::Arc::new(CapturingRecorder::default()); + let exec = ShadowBgpExecutor::new(rec.clone()); + exec.announce(route("203.0.113.7")).await.unwrap(); + let captured = rec.0.lock().unwrap(); + assert_eq!(captured.len(), 1); + assert!(matches!(captured[0], ShadowAction::RtbhAnnounce(_))); + // PanicExecutor is never constructed into the executor -> proves no BGP path. + } + + #[tokio::test] + async fn withdraw_and_flowspec_record_the_right_variants() { + let rec = std::sync::Arc::new(CapturingRecorder::default()); + let exec = ShadowBgpExecutor::new(rec.clone()); + exec.withdraw("203.0.113.7/32".parse().unwrap()) + .await + .unwrap(); + exec.announce_flowspec(flowspec_rule()).await.unwrap(); + let c = rec.0.lock().unwrap(); + assert!(matches!(c[0], ShadowAction::RtbhWithdraw(_))); + assert!(matches!(c[1], ShadowAction::FlowSpecAnnounce(_))); + } + + #[tokio::test] + async fn withdraw_flowspec_records_the_right_variant() { + let rec = std::sync::Arc::new(CapturingRecorder::default()); + let exec = ShadowBgpExecutor::new(rec.clone()); + exec.withdraw_flowspec(flowspec_rule()).await.unwrap(); + let c = rec.0.lock().unwrap(); + assert!(matches!(c[0], ShadowAction::FlowSpecWithdraw(_))); + } + + #[tokio::test] + async fn no_op_journal_never_errors() { + let j = NoOpJournal; + BlackholeJournal::record_announce( + &j, + "203.0.113.7".parse().unwrap(), + BlackholeOrigin::Auto, + 0, + ) + .await + .unwrap(); + BlackholeJournal::record_withdraw(&j, "203.0.113.7".parse().unwrap(), 0) + .await + .unwrap(); + FlowSpecJournal::record_announce(&j, flowspec_rule(), BlackholeOrigin::Auto, 0) + .await + .unwrap(); + FlowSpecJournal::record_withdraw(&j, flowspec_rule(), 0) + .await + .unwrap(); + } +} From f3aa0b1c3bf78401d0444e42334fcad7b2b14414 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 14:57:39 -0400 Subject: [PATCH 2/6] =?UTF-8?q?feat(config):=20shadow=20directive=20?= =?UTF-8?q?=E2=86=92=20Policy.shadow=20(opt-in,=20default=20live)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/blackwall-config/src/parser.rs | 29 +++++++++++++++++++++ crates/blackwall-core/src/policy.rs | 3 +++ crates/blackwall-core/src/resolve.rs | 2 ++ crates/blackwall-deception/tests/interop.rs | 4 +++ crates/blackwall-discovery/src/reconcile.rs | 1 + crates/blackwall-nft/src/render.rs | 3 +++ crates/blackwall-nft/tests/apply_netns.rs | 2 ++ crates/blackwall-state/src/lib.rs | 2 ++ 8 files changed, 46 insertions(+) diff --git a/crates/blackwall-config/src/parser.rs b/crates/blackwall-config/src/parser.rs index 51da544..fde0305 100644 --- a/crates/blackwall-config/src/parser.rs +++ b/crates/blackwall-config/src/parser.rs @@ -28,6 +28,7 @@ pub fn parse(lines: &[Line]) -> Result { let mut xdp: Option = None; let mut stateless_tcp_ports: Vec = Vec::new(); let mut pops: Vec = Vec::new(); + let mut shadow = false; let mut i = 0; while i < lines.len() { @@ -772,6 +773,16 @@ pub fn parse(lines: &[Line]) -> Result { } } } + "shadow" => { + if line.words.len() > 1 { + return Err(ConfigError::BadValue { + line: line.number, + what: "shadow", + value: line.words[1..].join(" "), + }); + } + shadow = true; + } other => { return Err(ConfigError::UnknownDirective { line: line.number, @@ -816,6 +827,7 @@ pub fn parse(lines: &[Line]) -> Result { flowtable, xdp, stateless_tcp_ports, + shadow, }) } @@ -1688,6 +1700,23 @@ flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s bogus= assert_eq!(p.metrics_listen, None); } + #[test] + fn parses_shadow_directive() { + let p = parse_text("interface wan eth0\nshadow\n").unwrap(); + assert!(p.shadow); + } + + #[test] + fn shadow_defaults_false() { + let p = parse_text("interface wan eth0\n").unwrap(); + assert!(!p.shadow); + } + + #[test] + fn shadow_rejects_trailing_tokens() { + assert!(parse_text("interface wan eth0\nshadow rtbh\n").is_err()); + } + #[test] fn parses_api_directive() { let p = parse_text( diff --git a/crates/blackwall-core/src/policy.rs b/crates/blackwall-core/src/policy.rs index ea71d07..caa9e16 100644 --- a/crates/blackwall-core/src/policy.rs +++ b/crates/blackwall-core/src/policy.rs @@ -74,4 +74,7 @@ pub struct Policy { /// Set via the `stateless-tcp ports=` directive; empty (the default) /// preserves today's behaviour where all deception TCP is interactive. pub stateless_tcp_ports: Vec, + /// Shadow mode: log + record + meter mitigations (RTBH/FlowSpec/XDP) + /// without applying them. `false` (the default) is live. + pub shadow: bool, } diff --git a/crates/blackwall-core/src/resolve.rs b/crates/blackwall-core/src/resolve.rs index 5cf5603..2ab78ab 100644 --- a/crates/blackwall-core/src/resolve.rs +++ b/crates/blackwall-core/src/resolve.rs @@ -139,6 +139,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } @@ -304,6 +305,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, }; let resolved = policy.resolve().expect("empty policy resolves"); assert!(resolved.is_empty()); diff --git a/crates/blackwall-deception/tests/interop.rs b/crates/blackwall-deception/tests/interop.rs index 64b1ef0..d344ea0 100644 --- a/crates/blackwall-deception/tests/interop.rs +++ b/crates/blackwall-deception/tests/interop.rs @@ -123,6 +123,7 @@ async fn serves_deception_banner() { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, }; // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000 @@ -189,6 +190,7 @@ async fn serves_deception_under_load() { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, }; // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000 @@ -277,6 +279,7 @@ fn serves_stateless_syn_cookie() { // The stateless-tier port under test (Component 2c wiring): deception // TCP on 8080 is routed to the engine's NFQUEUE instead of tproxy. stateless_tcp_ports: vec![8080], + shadow: false, }; // Apply the REAL nft ruleset: stateless-tcp TCP on 8080 -> nfqueue @@ -361,6 +364,7 @@ fn serves_stateless_syn_cookie_v6() { flowtable: None, xdp: None, stateless_tcp_ports: vec![8080], + shadow: false, }; blackwall_nft::apply(&policy).expect("nft apply"); diff --git a/crates/blackwall-discovery/src/reconcile.rs b/crates/blackwall-discovery/src/reconcile.rs index baf8e88..394eede 100644 --- a/crates/blackwall-discovery/src/reconcile.rs +++ b/crates/blackwall-discovery/src/reconcile.rs @@ -131,6 +131,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } diff --git a/crates/blackwall-nft/src/render.rs b/crates/blackwall-nft/src/render.rs index 2830851..1ee81d7 100644 --- a/crates/blackwall-nft/src/render.rs +++ b/crates/blackwall-nft/src/render.rs @@ -722,6 +722,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } @@ -752,6 +753,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } @@ -1258,6 +1260,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, }; let ruleset = render(&policy).expect("render empty"); // No resolved services, so real_v4 and real_v6 sets are empty. diff --git a/crates/blackwall-nft/tests/apply_netns.rs b/crates/blackwall-nft/tests/apply_netns.rs index 4b58fb7..9d758c8 100644 --- a/crates/blackwall-nft/tests/apply_netns.rs +++ b/crates/blackwall-nft/tests/apply_netns.rs @@ -35,6 +35,7 @@ fn sample() -> Policy { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } @@ -92,6 +93,7 @@ fn stale_set_elements_removed_on_second_apply() { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, }; blackwall_nft::apply(&policy_empty).expect("second apply"); diff --git a/crates/blackwall-state/src/lib.rs b/crates/blackwall-state/src/lib.rs index 338ba0f..3aed00e 100644 --- a/crates/blackwall-state/src/lib.rs +++ b/crates/blackwall-state/src/lib.rs @@ -1871,6 +1871,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } @@ -1909,6 +1910,7 @@ mod tests { flowtable: None, xdp: None, stateless_tcp_ports: Vec::new(), + shadow: false, } } From 253a3f09170c80e23933e11944b690b3f6116698 Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 15:10:26 -0400 Subject: [PATCH 3/6] feat(shadow): audit recorder + metric + shadow wiring for RTBH/FlowSpec --- bin/blackwalld/src/main.rs | 244 ++++++++++++++++++---------- bin/blackwalld/src/metrics.rs | 41 +++++ bin/blackwalld/src/shadow.rs | 90 ++++++++++ crates/blackwall-state/src/audit.rs | 5 +- crates/blackwall-state/src/lib.rs | 49 ++++++ scripts/coverage.sh | 3 +- 6 files changed, 345 insertions(+), 87 deletions(-) create mode 100644 bin/blackwalld/src/shadow.rs diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs index e9ecec7..dd8755e 100644 --- a/bin/blackwalld/src/main.rs +++ b/bin/blackwalld/src/main.rs @@ -2,6 +2,7 @@ mod api; mod metrics; +mod shadow; use clap::{Parser, Subcommand}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; @@ -770,11 +771,14 @@ async fn bgp_supervisor(mut states: tokio::sync::watch::Receiver, +async fn rtbh_manager_task( + mut manager: blackwall_rtbh::RtbhManager, mut rx: mpsc::Receiver, request_store: std::sync::Arc, -) { +) where + B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, + J: blackwall_rtbh::manager::BlackholeJournal + Send + 'static, +{ let mut ticker = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tokio::select! { @@ -814,11 +818,14 @@ async fn rtbh_manager_task( /// still-pending `add` for the same target (the operator's remove is the /// newer intent and must win over a not-yet-applied add), then marks this /// row `applied`. -async fn apply_request( - manager: &mut blackwall_rtbh::RtbhManager, +async fn apply_request( + manager: &mut blackwall_rtbh::RtbhManager, request_store: &blackwall_state::Store, req: blackwall_state::RtbhRequestRow, -) { +) where + B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, + J: blackwall_rtbh::manager::BlackholeJournal + Send + 'static, +{ match req.action.as_str() { "add" => match manager.apply_add(req.target, mono_now(), wall_now()).await { blackwall_rtbh::ApplyOutcome::Applied => { @@ -878,11 +885,14 @@ async fn apply_request( /// /// Runs until `rx` is closed (i.e. for the process's lifetime, since the /// paired `SelectorSink`'s sender is held by the running collector). -async fn flowspec_manager_task( - mut manager: blackwall_rtbh::FlowSpecManager, +async fn flowspec_manager_task( + mut manager: blackwall_rtbh::FlowSpecManager, mut rx: mpsc::Receiver, request_store: std::sync::Arc, -) { +) where + B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, + J: blackwall_rtbh::flowspec_manager::FlowSpecJournal + Send + 'static, +{ let mut ticker = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tokio::select! { @@ -930,11 +940,14 @@ async fn flowspec_manager_task( /// `pending` (retried on the next tick); `Rejected` marks it `rejected` with /// the reason. For `"remove"`: withdraws the flow, supersedes any other /// still-pending `add` for the same flow key, then marks this row `applied`. -async fn apply_flowspec_request( - manager: &mut blackwall_rtbh::FlowSpecManager, +async fn apply_flowspec_request( + manager: &mut blackwall_rtbh::FlowSpecManager, request_store: &blackwall_state::Store, req: blackwall_state::FlowSpecRequestRow, -) { +) where + B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, + J: blackwall_rtbh::flowspec_manager::FlowSpecJournal + Send + 'static, +{ let rule = blackwall_bgp::FlowSpecRule { dst: host_prefix(req.dst), protocol: Some(req.proto), @@ -1322,6 +1335,11 @@ async fn run() -> Result<(), Box> { hold_down_secs, } => { let policy = blackwall_config::parse_file(&config)?; + if policy.shadow { + tracing::warn!( + "SHADOW MODE — mitigations are LOGGED, NOT APPLIED (RTBH/FlowSpec/XDP)" + ); + } let database_url = std::env::var("DATABASE_URL") .map_err(|_| "DATABASE_URL must be set for the flow detector")?; let store = std::sync::Arc::new(blackwall_state::Store::connect(&database_url).await?); @@ -1346,6 +1364,11 @@ async fn run() -> Result<(), Box> { std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); // Captured inside the rtbh arm below so /metrics can report session state. let mut bgp_for_metrics: Option = None; + // Shared shadow-mode counters: fed by the RTBH/FlowSpec managers + // below (when `policy.shadow`) and by the XDP shadow gate, read by + // the metrics endpoint. Built unconditionally — harmless all-zero + // counters when shadow mode is off. + let shadow_metrics = std::sync::Arc::new(shadow::ShadowMetrics::default()); let sink: std::sync::Arc = match policy.rtbh.clone() { @@ -1354,47 +1377,66 @@ async fn run() -> Result<(), Box> { let pg_sink: std::sync::Arc = std::sync::Arc::new(blackwall_state::PgMitigationSink::new(store.clone())); - let peer = blackwall_bgp::PeerConfig { - local_asn: rtbh.local_asn, - peer_asn: rtbh.peer_asn, - peer_addr: rtbh.peer_addr, - router_id: rtbh.router_id, - hold_time: 90, - md5: rtbh.md5.as_ref().map(|s| s.reveal().to_owned()), - gtsm_hops: rtbh.gtsm_hops, - }; - // `BgpHandle` is a cloneable mpsc sender; both the RTBH and - // (optionally) FlowSpec managers share the one iBGP session. - let (bgp, _bgp_join) = blackwall_bgp::spawn(peer)?; - // Supervise the session: log loudly when it leaves Established - // (mitigations aren't reaching the peer) — issue #79. - tokio::spawn(bgp_supervisor(bgp.state_watch())); - bgp_for_metrics = Some(bgp.clone()); let controller = blackwall_rtbh::RtbhController::new(rtbh_config_from(&policy, &rtbh)); - let journal: blackwall_state::Store = (*store).clone(); - let mut manager = - blackwall_rtbh::RtbhManager::new(controller, bgp.clone(), journal); - - // Rehydrate the controller from the announced mirror before - // this session starts accepting new detections/requests. - let mirror = store.list_active_blackholes().await?; - let rehydrate_rows: Vec<(IpAddr, u64, blackwall_rtbh::BlackholeOrigin)> = - mirror - .into_iter() - .map(|row| { - let origin = match row.origin.as_str() { - "manual" => blackwall_rtbh::BlackholeOrigin::Manual, - _ => blackwall_rtbh::BlackholeOrigin::Auto, - }; - (row.target, row.announced_at_ms, origin) - }) - .collect(); - manager.rehydrate(rehydrate_rows, mono_now()).await; - let channel_cap = rtbh.max_blackholes.max(1024); let (tx, rx) = mpsc::channel::(channel_cap); - tokio::spawn(rtbh_manager_task(manager, rx, store.clone())); + + if policy.shadow { + let recorder = + shadow::AuditShadowRecorder::new(store.clone(), shadow_metrics.clone()); + let exec = blackwall_rtbh::ShadowBgpExecutor::new(recorder); + let manager = blackwall_rtbh::RtbhManager::new( + controller, + exec, + blackwall_rtbh::NoOpJournal, + ); + // No rehydrate: the shadow mirror is intentionally + // empty — nothing was ever really announced, so there + // is nothing to replay. + tokio::spawn(rtbh_manager_task(manager, rx, store.clone())); + } else { + let peer = blackwall_bgp::PeerConfig { + local_asn: rtbh.local_asn, + peer_asn: rtbh.peer_asn, + peer_addr: rtbh.peer_addr, + router_id: rtbh.router_id, + hold_time: 90, + md5: rtbh.md5.as_ref().map(|s| s.reveal().to_owned()), + gtsm_hops: rtbh.gtsm_hops, + }; + // `BgpHandle` is a cloneable mpsc sender; both the RTBH + // and (optionally) FlowSpec managers share the one + // iBGP session. + let (bgp, _bgp_join) = blackwall_bgp::spawn(peer)?; + // Supervise the session: log loudly when it leaves + // Established (mitigations aren't reaching the peer) + // — issue #79. + tokio::spawn(bgp_supervisor(bgp.state_watch())); + bgp_for_metrics = Some(bgp.clone()); + let journal: blackwall_state::Store = (*store).clone(); + let mut manager = + blackwall_rtbh::RtbhManager::new(controller, bgp.clone(), journal); + + // Rehydrate the controller from the announced mirror + // before this session starts accepting new + // detections/requests. + let mirror = store.list_active_blackholes().await?; + let rehydrate_rows: Vec<(IpAddr, u64, blackwall_rtbh::BlackholeOrigin)> = + mirror + .into_iter() + .map(|row| { + let origin = match row.origin.as_str() { + "manual" => blackwall_rtbh::BlackholeOrigin::Manual, + _ => blackwall_rtbh::BlackholeOrigin::Auto, + }; + (row.target, row.announced_at_ms, origin) + }) + .collect(); + manager.rehydrate(rehydrate_rows, mono_now()).await; + + tokio::spawn(rtbh_manager_task(manager, rx, store.clone())); + } match policy.flowspec.clone() { // RTBH-only: today's behaviour, Fanout([Pg, Channel→rtbh]). @@ -1406,48 +1448,80 @@ async fn run() -> Result<(), Box> { channel_sink, ])) } - // RTBH + FlowSpec: build a second single-owner manager off - // the SAME BGP session and route detections through a - // SelectorSink instead of the plain RTBH ChannelSink. + // RTBH + FlowSpec: build a second single-owner manager + // (shadow or, off the SAME live BGP session, real) and + // route detections through a SelectorSink instead of + // the plain RTBH ChannelSink. Some(fs) => { let fs_controller = blackwall_rtbh::FlowSpecController::new( flowspec_config_from(&policy, &fs), ); - let fs_journal: blackwall_state::Store = (*store).clone(); - let mut fs_manager = blackwall_rtbh::FlowSpecManager::new( - fs_controller, - bgp, - fs_journal, - ); - - // Rehydrate FlowSpec rules from the announced mirror. - let fs_mirror = store.list_active_flowspec().await?; - let fs_rehydrate: Vec<( - blackwall_bgp::FlowSpecRule, - u64, - blackwall_rtbh::BlackholeOrigin, - )> = fs_mirror - .into_iter() - .map(|row| { - let origin = match row.origin.as_str() { - "manual" => blackwall_rtbh::BlackholeOrigin::Manual, - _ => blackwall_rtbh::BlackholeOrigin::Auto, - }; - let rule = blackwall_bgp::FlowSpecRule { - dst: host_prefix(row.dst), - protocol: Some(row.proto), - dst_port: Some(row.dst_port), - action: blackwall_bgp::FlowAction::TrafficRate(row.rate), - }; - (rule, row.announced_at_ms, origin) - }) - .collect(); - fs_manager.rehydrate(fs_rehydrate, mono_now()).await; - let fs_cap = fs.max_rules.max(1024); let (fs_tx, fs_rx) = mpsc::channel::(fs_cap); - tokio::spawn(flowspec_manager_task(fs_manager, fs_rx, store.clone())); + + if policy.shadow { + let recorder = shadow::AuditShadowRecorder::new( + store.clone(), + shadow_metrics.clone(), + ); + let exec = blackwall_rtbh::ShadowBgpExecutor::new(recorder); + let fs_manager = blackwall_rtbh::FlowSpecManager::new( + fs_controller, + exec, + blackwall_rtbh::NoOpJournal, + ); + // No rehydrate: shadow mirror stays empty. + tokio::spawn(flowspec_manager_task( + fs_manager, + fs_rx, + store.clone(), + )); + } else { + // The live RTBH branch above always sets + // `bgp_for_metrics` before this point. + let bgp = bgp_for_metrics.clone().expect( + "live path sets bgp_for_metrics before FlowSpec construction", + ); + let fs_journal: blackwall_state::Store = (*store).clone(); + let mut fs_manager = blackwall_rtbh::FlowSpecManager::new( + fs_controller, + bgp, + fs_journal, + ); + + // Rehydrate FlowSpec rules from the announced mirror. + let fs_mirror = store.list_active_flowspec().await?; + let fs_rehydrate: Vec<( + blackwall_bgp::FlowSpecRule, + u64, + blackwall_rtbh::BlackholeOrigin, + )> = fs_mirror + .into_iter() + .map(|row| { + let origin = match row.origin.as_str() { + "manual" => blackwall_rtbh::BlackholeOrigin::Manual, + _ => blackwall_rtbh::BlackholeOrigin::Auto, + }; + let rule = blackwall_bgp::FlowSpecRule { + dst: host_prefix(row.dst), + protocol: Some(row.proto), + dst_port: Some(row.dst_port), + action: blackwall_bgp::FlowAction::TrafficRate( + row.rate, + ), + }; + (rule, row.announced_at_ms, origin) + }) + .collect(); + fs_manager.rehydrate(fs_rehydrate, mono_now()).await; + + tokio::spawn(flowspec_manager_task( + fs_manager, + fs_rx, + store.clone(), + )); + } let selection = blackwall_flow::SelectionConfig { concentration: fs.concentration, @@ -1649,6 +1723,7 @@ async fn run() -> Result<(), Box> { stateless: None, afxdp_udp_responses: afxdp_udp_metric.clone(), agent_stats: Some(agent_snapshot.clone()), + shadow: Some(shadow_metrics.clone()), }; tokio::spawn(metrics::metrics_server(metrics_listen, sources)); } @@ -1901,6 +1976,7 @@ async fn run() -> Result<(), Box> { stateless: Some(stateless_metrics.clone()), afxdp_udp_responses: None, agent_stats: None, + shadow: None, }; tokio::spawn(metrics::metrics_server(metrics_listen, sources)); } diff --git a/bin/blackwalld/src/metrics.rs b/bin/blackwalld/src/metrics.rs index 1fe829f..6c9b509 100644 --- a/bin/blackwalld/src/metrics.rs +++ b/bin/blackwalld/src/metrics.rs @@ -44,6 +44,9 @@ pub(crate) struct MetricsSources { /// Per-POP agent telemetry snapshot, refreshed once per collector tick; /// `None` outside the flow daemon (no sFlow collector, no agents). pub agent_stats: Option>>>, + /// Shadow-mode "would mitigate" counters (RTBH/FlowSpec/XDP); `None` + /// outside the flow daemon (no RTBH/FlowSpec/XDP managers to shadow). + pub shadow: Option>, } /// Correctly-rounded `u64 -> f64` without an `as` cast: `u32 -> f64` is exact @@ -313,6 +316,38 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option { Some(out) } +/// Render `blackwall_shadow_would_mitigate_total{plane,action}` from the +/// shared shadow counters, or `None` when shadow counters aren't wired up +/// (outside the flow daemon). Labels are a fixed, known-at-compile-time set +/// (unlike [`agent_stats_block`]'s dynamic POP names), but still hand-written +/// since [`Metric`] only carries unlabelled series. +fn shadow_block(sources: &MetricsSources) -> Option { + use std::sync::atomic::Ordering; + + let shadow = sources.shadow.as_ref()?; + let mut out = String::new(); + let _ = writeln!( + out, + "# HELP blackwall_shadow_would_mitigate_total Mitigations that would have been applied under shadow mode, by plane and action" + ); + let _ = writeln!(out, "# TYPE blackwall_shadow_would_mitigate_total counter"); + for (plane, action, counter) in [ + ("rtbh", "announce", &shadow.rtbh_announce), + ("rtbh", "withdraw", &shadow.rtbh_withdraw), + ("flowspec", "announce", &shadow.flowspec_announce), + ("flowspec", "withdraw", &shadow.flowspec_withdraw), + ("xdp", "block", &shadow.xdp_block), + ("xdp", "rate_limit", &shadow.xdp_rate_limit), + ] { + let _ = writeln!( + out, + "blackwall_shadow_would_mitigate_total{{plane=\"{plane}\",action=\"{action}\"}} {}", + counter.load(Ordering::Relaxed) + ); + } + Some(out) +} + /// Serve `/metrics` forever. Each connection is handled on its own task so a /// slow client cannot block scrapes; a bind failure disables the endpoint (and /// is logged) without taking down the daemon. @@ -360,6 +395,12 @@ async fn handle_conn(mut sock: tokio::net::TcpStream, sources: &MetricsSources) } body.push_str(&agent); } + if let Some(shadow) = shadow_block(sources) { + if !body.is_empty() { + body.push('\n'); + } + body.push_str(&shadow); + } format!( "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len() diff --git a/bin/blackwalld/src/shadow.rs b/bin/blackwalld/src/shadow.rs new file mode 100644 index 0000000..fb09f53 --- /dev/null +++ b/bin/blackwalld/src/shadow.rs @@ -0,0 +1,90 @@ +//! Concrete shadow recorder: logs, meters, and audit-logs intended mitigations. +//! I/O glue — coverage-excluded. + +use blackwall_rtbh::{ShadowAction, ShadowRecorder}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +/// Per-(plane, action) counters backing the `blackwall_shadow_would_mitigate_total` +/// metric. Shared (behind an `Arc`) across the RTBH manager, the FlowSpec +/// manager, the XDP shadow gate, and the `/metrics` endpoint. +#[derive(Default)] +pub struct ShadowMetrics { + /// RTBH blackhole announcements that would have been sent. + pub rtbh_announce: AtomicU64, + /// RTBH blackhole withdrawals that would have been sent. + pub rtbh_withdraw: AtomicU64, + /// FlowSpec rule announcements that would have been sent. + pub flowspec_announce: AtomicU64, + /// FlowSpec rule withdrawals that would have been sent. + pub flowspec_withdraw: AtomicU64, + /// XDP blocks that would have been installed. + pub xdp_block: AtomicU64, + /// XDP rate limits that would have been installed. + pub xdp_rate_limit: AtomicU64, +} + +/// Records shadow actions to the audit log + metrics + INFO log. +/// +/// Wired in place of a real `BgpExecutor`/journal pair (via +/// [`blackwall_rtbh::ShadowBgpExecutor`]) when the `shadow` config directive +/// is set: every mitigation the manager would have applied is logged, +/// counted, and durably audited instead of reaching a real BGP session. +pub struct AuditShadowRecorder { + store: Arc, + metrics: Arc, +} + +impl AuditShadowRecorder { + /// Build a recorder that audits to `store` and meters into `metrics`. + pub fn new(store: Arc, metrics: Arc) -> Self { + Self { store, metrics } + } +} + +#[async_trait::async_trait] +impl ShadowRecorder for AuditShadowRecorder { + async fn record(&self, action: ShadowAction) { + let (plane, verb, target, counter): (&str, &str, String, &AtomicU64) = match &action { + ShadowAction::RtbhAnnounce(r) => ( + "rtbh", + "announce", + format!("{r:?}"), + &self.metrics.rtbh_announce, + ), + ShadowAction::RtbhWithdraw(p) => ( + "rtbh", + "withdraw", + p.to_string(), + &self.metrics.rtbh_withdraw, + ), + ShadowAction::FlowSpecAnnounce(r) => ( + "flowspec", + "announce", + format!("{r:?}"), + &self.metrics.flowspec_announce, + ), + ShadowAction::FlowSpecWithdraw(r) => ( + "flowspec", + "withdraw", + format!("{r:?}"), + &self.metrics.flowspec_withdraw, + ), + }; + counter.fetch_add(1, Ordering::Relaxed); + tracing::info!( + plane, + verb, + target = %target, + "shadow: would mitigate (logged, not applied)" + ); + let detail = serde_json::json!({ "plane": plane, "verb": verb, "target": target }); + if let Err(err) = self + .store + .record_audit("shadow", &format!("shadow.{plane}.{verb}"), &detail) + .await + { + tracing::warn!(%err, "shadow: audit write failed (mitigation still suppressed)"); + } + } +} diff --git a/crates/blackwall-state/src/audit.rs b/crates/blackwall-state/src/audit.rs index d7c1775..83a4877 100644 --- a/crates/blackwall-state/src/audit.rs +++ b/crates/blackwall-state/src/audit.rs @@ -1,7 +1,8 @@ //! Audit-log helpers. -// The append happens inside `Store::apply_policy`'s transaction (lib.rs). -// `Store::audit_count` (lib.rs) is the first read accessor; richer queries +// Appends happen inside `Store::apply_policy`'s transaction, or standalone +// via `Store::record_audit` (both in lib.rs). `Store::audit_count` and +// `Store::list_recent_audit` (lib.rs) are the read accessors; richer queries // land here as the API grows. /// One `audit_log` row. diff --git a/crates/blackwall-state/src/lib.rs b/crates/blackwall-state/src/lib.rs index 3aed00e..53ee546 100644 --- a/crates/blackwall-state/src/lib.rs +++ b/crates/blackwall-state/src/lib.rs @@ -305,6 +305,26 @@ impl Store { Ok(out) } + /// Append one row to the `audit_log`: `actor` performed `action`, with + /// structured `detail`. A standalone counterpart to the audit insert + /// embedded in [`Store::apply_policy`]'s transaction, for call sites + /// (e.g. shadow-mode mitigation recording) that have no transaction of + /// their own and just need a single durable audit row. + pub async fn record_audit( + &self, + actor: &str, + action: &str, + detail: &serde_json::Value, + ) -> Result<(), StateError> { + sqlx::query("INSERT INTO audit_log (actor, action, detail) VALUES ($1, $2, $3)") + .bind(actor) + .bind(action) + .bind(detail) + .execute(&self.pool) + .await?; + Ok(()) + } + /// Count audit-log entries. pub async fn audit_count(&self) -> Result { let row: (i64,) = sqlx::query_as("SELECT count(*) FROM audit_log") @@ -2066,4 +2086,33 @@ mod tests { "one of our own apply_policy calls must appear among the 10 most recent audit rows" ); } + + #[tokio::test] + async fn record_audit_appends_a_row_with_the_given_fields() { + let Some(url) = test_url() else { + eprintln!("DATABASE_URL not set; skipping"); + return; + }; + let store = Store::connect(&url).await.expect("connect"); + store.migrate().await.expect("migrate"); + let _guard = DB_SERIAL_LOCK.lock().await; + + let before = store.audit_count().await.expect("count before"); + let detail = serde_json::json!({ "plane": "rtbh", "verb": "announce" }); + store + .record_audit("shadow", "shadow.rtbh.announce", &detail) + .await + .expect("record_audit"); + let after = store.audit_count().await.expect("count after"); + assert_eq!( + after, + before + 1, + "record_audit must append exactly one row" + ); + + let recent = store.list_recent_audit(1).await.expect("list limit 1"); + assert_eq!(recent[0].actor, "shadow"); + assert_eq!(recent[0].action, "shadow.rtbh.announce"); + assert_eq!(recent[0].detail, detail); + } } diff --git a/scripts/coverage.sh b/scripts/coverage.sh index 3bf9753..3c518d0 100755 --- a/scripts/coverage.sh +++ b/scripts/coverage.sh @@ -7,6 +7,7 @@ # - transport/{tproxy,nfqueue}.rs TPROXY transparent sockets, NFQUEUE + raw sockets # - blackwall-nft/src/apply.rs nftables kernel apply # - blackwalld/src/main.rs daemon process/runtime glue +# - blackwalld/src/shadow.rs shadow-mode audit/log/meter I/O glue # - discovery/src/{incus_client,proc_io}.rs Incus unix-socket + /proc readers # - speedtest/src/providers/*_net.rs live HTTP/TCP speedtest fetchers # - blackwall-lab/src/{exec/*,cli,bin/lab}.rs netns/process orchestration (needs CAP_NET_ADMIN) @@ -28,6 +29,6 @@ # Extra args are forwarded to cargo llvm-cov (e.g. --html, --summary-only). set -euo pipefail -EXCLUDE='(transport/(tproxy|nfqueue)\.rs|blackwall-nft/src/apply\.rs|blackwalld/src/(main|metrics|api)\.rs|discovery/src/incus_client\.rs|discovery/src/proc_io\.rs|speedtest/src/providers/.*_net\.rs|shaper/src/apply\.rs|dns/src/send_net\.rs|flow/src/collector_net\.rs|bgp/src/session_net\.rs|blackwall-lab/src/exec/.*\.rs|blackwall-lab/src/cli\.rs|blackwall-lab/src/bin/lab\.rs|blackwall-trafficgen/src/io/.*\.rs|blackwall-trafficgen/src/bin/.*\.rs|blackwall-xdp/src/dataplane\.rs|blackwall-xdp/src/afxdp\.rs|blackwall-xdp/src/capture\.rs)' +EXCLUDE='(transport/(tproxy|nfqueue)\.rs|blackwall-nft/src/apply\.rs|blackwalld/src/(main|metrics|api|shadow)\.rs|discovery/src/incus_client\.rs|discovery/src/proc_io\.rs|speedtest/src/providers/.*_net\.rs|shaper/src/apply\.rs|dns/src/send_net\.rs|flow/src/collector_net\.rs|bgp/src/session_net\.rs|blackwall-lab/src/exec/.*\.rs|blackwall-lab/src/cli\.rs|blackwall-lab/src/bin/lab\.rs|blackwall-trafficgen/src/io/.*\.rs|blackwall-trafficgen/src/bin/.*\.rs|blackwall-xdp/src/dataplane\.rs|blackwall-xdp/src/afxdp\.rs|blackwall-xdp/src/capture\.rs)' exec cargo llvm-cov --workspace --fail-under-lines 90 --ignore-filename-regex "$EXCLUDE" "$@" From b1e56c2fd036012465c3c0a464dbf63fc4a085fd Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 15:18:08 -0400 Subject: [PATCH 4/6] =?UTF-8?q?feat(shadow):=20gate=20XDP=20map-apply=20?= =?UTF-8?q?=E2=80=94=20record=20intended=20drops,=20don't=20write=20maps?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bin/blackwalld/src/main.rs | 32 ++++++---- bin/blackwalld/src/shadow.rs | 115 +++++++++++++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 11 deletions(-) diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs index dd8755e..e84720a 100644 --- a/bin/blackwalld/src/main.rs +++ b/bin/blackwalld/src/main.rs @@ -1009,12 +1009,11 @@ async fn apply_flowspec_request( /// controller can never overflow a map. const XDP_MAX_ENTRIES: usize = 65_536; -/// The concrete [`blackwall_xdp::manager::XdpManager`] the daemon runs: a -/// shared attached data plane as executor and the Postgres journal as mirror. -type DaemonXdpManager = blackwall_xdp::manager::XdpManager< - std::sync::Arc, - blackwall_state::PgXdpJournal, ->; +/// The concrete [`blackwall_xdp::manager::XdpManager`] the daemon runs: an +/// executor that is either the live attached data plane or (in shadow mode) +/// [`shadow::XdpExec::Shadow`], plus the Postgres journal as mirror. +type DaemonXdpManager = + blackwall_xdp::manager::XdpManager; /// Build an [`ipnet::IpNet`] from a stored address + optional prefix length, /// falling back to a host route (`/32`/`/128`) when the length is absent. @@ -1675,11 +1674,22 @@ async fn run() -> Result<(), Box> { default_pps, ); let journal = blackwall_state::PgXdpJournal::new(store.clone()); - let mut manager = blackwall_xdp::manager::XdpManager::new( - controller, - dataplane.clone(), - journal, - ); + // Shadow mode: swap in `ShadowXdpExecutor` so every + // apply call site below (detections, manual CLI + // requests, restart rehydration) records + meters + // instead of writing the live eBPF maps — the maps + // stay untouched by this session. Live wiring + // (`XdpExec::Live`) is unchanged when `!policy.shadow`. + let executor = if policy.shadow { + shadow::XdpExec::Shadow(shadow::ShadowXdpExecutor::new( + store.clone(), + shadow_metrics.clone(), + )) + } else { + shadow::XdpExec::Live(dataplane.clone()) + }; + let mut manager = + blackwall_xdp::manager::XdpManager::new(controller, executor, journal); // Rehydrate the controller + maps from the active mirror // (blocks and rate limits, burst included) before this diff --git a/bin/blackwalld/src/shadow.rs b/bin/blackwalld/src/shadow.rs index fb09f53..cad3b1e 100644 --- a/bin/blackwalld/src/shadow.rs +++ b/bin/blackwalld/src/shadow.rs @@ -2,6 +2,7 @@ //! I/O glue — coverage-excluded. use blackwall_rtbh::{ShadowAction, ShadowRecorder}; +use blackwall_xdp::{XdpAction, XdpExecError, XdpExecutor}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -88,3 +89,117 @@ impl ShadowRecorder for AuditShadowRecorder { } } } + +/// [`XdpExecutor`] used in place of the live eBPF map writer when the +/// `shadow` config directive is set: install actions (`Block`/`RateLimit`) +/// are logged, metered into [`ShadowMetrics`], and audit-logged instead of +/// touching a map; removal actions (`Unblock`/`ClearRate`) are pure no-ops, +/// since shadow mode never installed anything for them to remove. +pub struct ShadowXdpExecutor { + store: Arc, + metrics: Arc, +} + +impl ShadowXdpExecutor { + /// Build an executor that audits to `store` and meters into `metrics`. + pub fn new(store: Arc, metrics: Arc) -> Self { + Self { store, metrics } + } +} + +#[async_trait::async_trait] +impl XdpExecutor for ShadowXdpExecutor { + async fn apply(&self, action: XdpAction) -> Result<(), XdpExecError> { + match action { + XdpAction::Block { net } => { + self.metrics.xdp_block.fetch_add(1, Ordering::Relaxed); + tracing::info!( + plane = "xdp", + verb = "block", + target = %net, + "shadow: would mitigate (not applied)" + ); + let detail = + serde_json::json!({"plane": "xdp", "verb": "block", "target": net.to_string()}); + if let Err(err) = self + .store + .record_audit("shadow", "shadow.xdp.block", &detail) + .await + { + tracing::warn!(%err, "shadow: audit write failed (mitigation still suppressed)"); + } + } + XdpAction::RateLimit { + src, + pps, + burst, + victim, + } => { + self.metrics.xdp_rate_limit.fetch_add(1, Ordering::Relaxed); + tracing::info!( + plane = "xdp", + verb = "rate_limit", + target = %src, + pps, + burst, + victim = victim.map(|v| v.to_string()), + "shadow: would mitigate (not applied)" + ); + let detail = serde_json::json!({ + "plane": "xdp", + "verb": "rate_limit", + "target": src.to_string(), + "pps": pps, + "burst": burst, + "victim": victim.map(|v| v.to_string()), + }); + if let Err(err) = self + .store + .record_audit("shadow", "shadow.xdp.rate_limit", &detail) + .await + { + tracing::warn!(%err, "shadow: audit write failed (mitigation still suppressed)"); + } + } + XdpAction::Unblock { net } => { + tracing::debug!( + plane = "xdp", + verb = "unblock", + target = %net, + "shadow: no-op (nothing was ever installed)" + ); + } + XdpAction::ClearRate { src } => { + tracing::debug!( + plane = "xdp", + verb = "clear_rate", + target = %src, + "shadow: no-op (nothing was ever installed)" + ); + } + } + Ok(()) + } +} + +/// Selects between the live eBPF map writer and [`ShadowXdpExecutor`] at +/// construction time, so [`crate::DaemonXdpManager`]'s single `XdpManager` +/// type serves both live and shadow sessions — every apply call site +/// (detections, manual CLI requests, restart rehydration) is gated by +/// whichever variant is installed, with no per-call-site branching. +pub enum XdpExec { + /// Writes straight to the live eBPF maps. + Live(Arc), + /// Shadow mode: records + meters, never touches a map. + Shadow(ShadowXdpExecutor), +} + +#[async_trait::async_trait] +impl XdpExecutor for XdpExec { + async fn apply(&self, action: XdpAction) -> Result<(), XdpExecError> { + match self { + Self::Live(dataplane) => dataplane.apply(action).await, + Self::Shadow(shadow) => shadow.apply(action).await, + } + } +} From d5e629c5084408b5731621651f1459a345c9e30b Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 15:26:12 -0400 Subject: [PATCH 5/6] fix(shadow): no-op XDP journal + skip rehydrate in shadow (keep mirror empty) --- bin/blackwalld/src/main.rs | 112 +++++++++++++++++----------- crates/blackwall-xdp/src/lib.rs | 3 +- crates/blackwall-xdp/src/manager.rs | 42 +++++++++++ 3 files changed, 113 insertions(+), 44 deletions(-) diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs index e84720a..e499fd0 100644 --- a/bin/blackwalld/src/main.rs +++ b/bin/blackwalld/src/main.rs @@ -1009,11 +1009,12 @@ async fn apply_flowspec_request( /// controller can never overflow a map. const XDP_MAX_ENTRIES: usize = 65_536; -/// The concrete [`blackwall_xdp::manager::XdpManager`] the daemon runs: an -/// executor that is either the live attached data plane or (in shadow mode) -/// [`shadow::XdpExec::Shadow`], plus the Postgres journal as mirror. -type DaemonXdpManager = - blackwall_xdp::manager::XdpManager; +/// The [`blackwall_xdp::manager::XdpManager`] the daemon runs: an executor +/// that is either the live attached data plane or (in shadow mode) +/// [`shadow::XdpExec::Shadow`], plus a journal `J` that is the Postgres +/// mirror ([`blackwall_state::PgXdpJournal`]) live, or the all-no-op +/// [`blackwall_xdp::NoOpXdpJournal`] in shadow (so the mirror stays empty). +type DaemonXdpManager = blackwall_xdp::manager::XdpManager; /// Build an [`ipnet::IpNet`] from a stored address + optional prefix length, /// falling back to a host route (`/32`/`/128`) when the length is absent. @@ -1069,12 +1070,14 @@ fn xdp_entry_to_action( /// When `auto_enabled` is false (no `default-rate-limit` configured) detection /// events are still drained off the channel but ignored — only operator CLI /// requests populate the maps. Runs until `rx` is closed. -async fn xdp_manager_task( - mut manager: DaemonXdpManager, +async fn xdp_manager_task( + mut manager: DaemonXdpManager, mut rx: mpsc::Receiver, request_store: std::sync::Arc, auto_enabled: bool, -) { +) where + J: blackwall_xdp::XdpJournal + 'static, +{ let mut ticker = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tokio::select! { @@ -1205,11 +1208,13 @@ fn afxdp_udp_responder_loop( /// it `pending` (retried on the next tick); `Rejected` marks it `rejected`. /// `unblock`/`clear_rate`: always applies, then marks `applied`. Unknown /// actions are logged and left untouched. -async fn apply_xdp_request( - manager: &mut DaemonXdpManager, +async fn apply_xdp_request( + manager: &mut DaemonXdpManager, request_store: &blackwall_state::Store, req: blackwall_state::XdpRequestRow, -) { +) where + J: blackwall_xdp::XdpJournal, +{ use blackwall_xdp::manager::ApplyOutcome; let mark = |id: i64, status: &'static str| async move { @@ -1673,43 +1678,64 @@ async fn run() -> Result<(), Box> { XDP_MAX_ENTRIES, default_pps, ); - let journal = blackwall_state::PgXdpJournal::new(store.clone()); - // Shadow mode: swap in `ShadowXdpExecutor` so every - // apply call site below (detections, manual CLI - // requests, restart rehydration) records + meters - // instead of writing the live eBPF maps — the maps - // stay untouched by this session. Live wiring - // (`XdpExec::Live`) is unchanged when `!policy.shadow`. - let executor = if policy.shadow { - shadow::XdpExec::Shadow(shadow::ShadowXdpExecutor::new( + let (xdp_tx, xdp_rx) = + mpsc::channel::(4096); + + // Shadow mode swaps BOTH I/O seams of the manager, so + // the session touches neither the eBPF maps nor the + // `xdp_entries` mirror: + // * executor → `ShadowXdpExecutor` (records + meters, + // never writes a map), and + // * journal → `NoOpXdpJournal` (persists nothing), + // and it SKIPS rehydrate — the shadow mirror is + // intentionally empty, so there is nothing to reapply + // (and reapplying would re-log stale rows as "would + // mitigate"). This mirrors the RTBH/FlowSpec shadow + // arms. When `!policy.shadow`, the live path + // (`XdpExec::Live` + `PgXdpJournal` + rehydrate) is + // exactly as before. + let handle = if policy.shadow { + let executor = shadow::XdpExec::Shadow(shadow::ShadowXdpExecutor::new( store.clone(), shadow_metrics.clone(), + )); + let manager = blackwall_xdp::manager::XdpManager::new( + controller, + executor, + blackwall_xdp::NoOpXdpJournal, + ); + // No rehydrate: the shadow mirror is intentionally empty. + tokio::spawn(xdp_manager_task( + manager, + xdp_rx, + store.clone(), + auto_enabled, )) } else { - shadow::XdpExec::Live(dataplane.clone()) - }; - let mut manager = - blackwall_xdp::manager::XdpManager::new(controller, executor, journal); - - // Rehydrate the controller + maps from the active mirror - // (blocks and rate limits, burst included) before this - // session accepts new detections/requests. - let rows: Vec<_> = store - .xdp_active() - .await? - .iter() - .filter_map(xdp_entry_to_action) - .collect(); - manager.reapply_active(rows).await; + let executor = shadow::XdpExec::Live(dataplane.clone()); + let journal = blackwall_state::PgXdpJournal::new(store.clone()); + let mut manager = blackwall_xdp::manager::XdpManager::new( + controller, executor, journal, + ); - let (xdp_tx, xdp_rx) = - mpsc::channel::(4096); - let handle = tokio::spawn(xdp_manager_task( - manager, - xdp_rx, - store.clone(), - auto_enabled, - )); + // Rehydrate the controller + maps from the active + // mirror (blocks and rate limits, burst included) + // before this session accepts new detections/requests. + let rows: Vec<_> = store + .xdp_active() + .await? + .iter() + .filter_map(xdp_entry_to_action) + .collect(); + manager.reapply_active(rows).await; + + tokio::spawn(xdp_manager_task( + manager, + xdp_rx, + store.clone(), + auto_enabled, + )) + }; xdp_shutdown = Some((handle, dataplane)); tracing::info!(interface = %iface, auto = auto_enabled, "XDP data plane attached"); diff --git a/crates/blackwall-xdp/src/lib.rs b/crates/blackwall-xdp/src/lib.rs index 6070691..3649875 100644 --- a/crates/blackwall-xdp/src/lib.rs +++ b/crates/blackwall-xdp/src/lib.rs @@ -13,7 +13,8 @@ pub use capture::{XdpCapture, DEFAULT_CAPTURE_PIN_DIR}; pub use control::{XdpAction, XdpController, XdpOrigin}; pub use dataplane::{XdpDataplane, XdpError, XdpStats}; pub use manager::{ - ApplyOutcome, XdpExecError, XdpExecutor, XdpJournal, XdpJournalError, XdpManager, + ApplyOutcome, NoOpXdpJournal, XdpExecError, XdpExecutor, XdpJournal, XdpJournalError, + XdpManager, }; pub use pcap::{to_pcap, CapturedPacket}; pub use sink::XdpMitigationSink; diff --git a/crates/blackwall-xdp/src/manager.rs b/crates/blackwall-xdp/src/manager.rs index 2271702..76265e8 100644 --- a/crates/blackwall-xdp/src/manager.rs +++ b/crates/blackwall-xdp/src/manager.rs @@ -47,6 +47,28 @@ pub struct XdpExecError; #[error("XDP journal error: {0}")] pub struct XdpJournalError(pub String); +/// An [`XdpJournal`] that persists nothing. +/// +/// Installed in place of the real persistence journal when the `shadow` +/// config directive is set, so the `xdp_entries` mirror stays empty: in +/// shadow mode no block or rate-limit is ever written to the eBPF maps, so +/// nothing must be journaled that a later live restart could rehydrate (via +/// [`XdpManager::reapply_active`]) and install for real. Mirrors +/// `blackwall_rtbh::NoOpJournal`. +pub struct NoOpXdpJournal; + +#[async_trait] +impl XdpJournal for NoOpXdpJournal { + async fn record( + &self, + _action: &XdpAction, + _origin: XdpOrigin, + _at_ms: u64, + ) -> Result<(), XdpJournalError> { + Ok(()) + } +} + /// Outcome of a manual [`XdpManager`] apply call. #[derive(Debug, PartialEq, Eq)] pub enum ApplyOutcome { @@ -581,4 +603,24 @@ mod tests { "repeated failures for one source coalesce to a single queued op" ); } + + #[tokio::test] + async fn noop_journal_records_nothing_and_succeeds() { + // The shadow-mode journal must accept every record call without error + // and persist nothing — it holds no state, so a `Block` and a + // `RateLimit` record both simply return Ok, leaving no observable + // mirror behind for a later live restart to rehydrate. + let journal = NoOpXdpJournal; + let block = XdpAction::Block { + net: "198.51.100.0/24".parse().unwrap(), + }; + let rate = XdpAction::RateLimit { + src: "198.51.100.9".parse().unwrap(), + pps: 500, + burst: 500, + victim: Some("203.0.113.7".parse().unwrap()), + }; + assert!(journal.record(&block, XdpOrigin::Manual, 0).await.is_ok()); + assert!(journal.record(&rate, XdpOrigin::Auto, 1000).await.is_ok()); + } } From fb8b2a7b35907ee4c1222d2bc40a4db5102f73bc Mon Sep 17 00:00:00 2001 From: Zoa Hickenlooper Date: Sat, 11 Jul 2026 15:37:27 -0400 Subject: [PATCH 6/6] fix(shadow): structured audit detail + harden live-bgp wiring --- bin/blackwalld/src/main.rs | 135 +++++++++++++++++++---------------- bin/blackwalld/src/shadow.rs | 57 +++++++++++++-- 2 files changed, 124 insertions(+), 68 deletions(-) diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs index e499fd0..ff6d8a6 100644 --- a/bin/blackwalld/src/main.rs +++ b/bin/blackwalld/src/main.rs @@ -1386,7 +1386,12 @@ async fn run() -> Result<(), Box> { let channel_cap = rtbh.max_blackholes.max(1024); let (tx, rx) = mpsc::channel::(channel_cap); - if policy.shadow { + // The live BGP handle, threaded to the FlowSpec + // construction below so its live branch reuses this same + // iBGP session. `None` in shadow mode (no real session is + // spawned) — which is exactly what the FlowSpec match keys + // off, so there is no implicit cross-block invariant. + let live_bgp: Option = if policy.shadow { let recorder = shadow::AuditShadowRecorder::new(store.clone(), shadow_metrics.clone()); let exec = blackwall_rtbh::ShadowBgpExecutor::new(recorder); @@ -1399,6 +1404,7 @@ async fn run() -> Result<(), Box> { // empty — nothing was ever really announced, so there // is nothing to replay. tokio::spawn(rtbh_manager_task(manager, rx, store.clone())); + None } else { let peer = blackwall_bgp::PeerConfig { local_asn: rtbh.local_asn, @@ -1440,7 +1446,8 @@ async fn run() -> Result<(), Box> { manager.rehydrate(rehydrate_rows, mono_now()).await; tokio::spawn(rtbh_manager_task(manager, rx, store.clone())); - } + Some(bgp) + }; match policy.flowspec.clone() { // RTBH-only: today's behaviour, Fanout([Pg, Channel→rtbh]). @@ -1464,67 +1471,69 @@ async fn run() -> Result<(), Box> { let (fs_tx, fs_rx) = mpsc::channel::(fs_cap); - if policy.shadow { - let recorder = shadow::AuditShadowRecorder::new( - store.clone(), - shadow_metrics.clone(), - ); - let exec = blackwall_rtbh::ShadowBgpExecutor::new(recorder); - let fs_manager = blackwall_rtbh::FlowSpecManager::new( - fs_controller, - exec, - blackwall_rtbh::NoOpJournal, - ); - // No rehydrate: shadow mirror stays empty. - tokio::spawn(flowspec_manager_task( - fs_manager, - fs_rx, - store.clone(), - )); - } else { - // The live RTBH branch above always sets - // `bgp_for_metrics` before this point. - let bgp = bgp_for_metrics.clone().expect( - "live path sets bgp_for_metrics before FlowSpec construction", - ); - let fs_journal: blackwall_state::Store = (*store).clone(); - let mut fs_manager = blackwall_rtbh::FlowSpecManager::new( - fs_controller, - bgp, - fs_journal, - ); - - // Rehydrate FlowSpec rules from the announced mirror. - let fs_mirror = store.list_active_flowspec().await?; - let fs_rehydrate: Vec<( - blackwall_bgp::FlowSpecRule, - u64, - blackwall_rtbh::BlackholeOrigin, - )> = fs_mirror - .into_iter() - .map(|row| { - let origin = match row.origin.as_str() { - "manual" => blackwall_rtbh::BlackholeOrigin::Manual, - _ => blackwall_rtbh::BlackholeOrigin::Auto, - }; - let rule = blackwall_bgp::FlowSpecRule { - dst: host_prefix(row.dst), - protocol: Some(row.proto), - dst_port: Some(row.dst_port), - action: blackwall_bgp::FlowAction::TrafficRate( - row.rate, - ), - }; - (rule, row.announced_at_ms, origin) - }) - .collect(); - fs_manager.rehydrate(fs_rehydrate, mono_now()).await; - - tokio::spawn(flowspec_manager_task( - fs_manager, - fs_rx, - store.clone(), - )); + // Reuse the live BGP handle from the RTBH branch + // (`Some` on the live path, `None` in shadow mode). + // Matching the real value here removes the earlier + // cross-block `.expect()` on `bgp_for_metrics`. + match live_bgp { + None => { + let recorder = shadow::AuditShadowRecorder::new( + store.clone(), + shadow_metrics.clone(), + ); + let exec = blackwall_rtbh::ShadowBgpExecutor::new(recorder); + let fs_manager = blackwall_rtbh::FlowSpecManager::new( + fs_controller, + exec, + blackwall_rtbh::NoOpJournal, + ); + // No rehydrate: shadow mirror stays empty. + tokio::spawn(flowspec_manager_task( + fs_manager, + fs_rx, + store.clone(), + )); + } + Some(bgp) => { + let fs_journal: blackwall_state::Store = (*store).clone(); + let mut fs_manager = blackwall_rtbh::FlowSpecManager::new( + fs_controller, + bgp, + fs_journal, + ); + + // Rehydrate FlowSpec rules from the announced mirror. + let fs_mirror = store.list_active_flowspec().await?; + let fs_rehydrate: Vec<( + blackwall_bgp::FlowSpecRule, + u64, + blackwall_rtbh::BlackholeOrigin, + )> = fs_mirror + .into_iter() + .map(|row| { + let origin = match row.origin.as_str() { + "manual" => blackwall_rtbh::BlackholeOrigin::Manual, + _ => blackwall_rtbh::BlackholeOrigin::Auto, + }; + let rule = blackwall_bgp::FlowSpecRule { + dst: host_prefix(row.dst), + protocol: Some(row.proto), + dst_port: Some(row.dst_port), + action: blackwall_bgp::FlowAction::TrafficRate( + row.rate, + ), + }; + (rule, row.announced_at_ms, origin) + }) + .collect(); + fs_manager.rehydrate(fs_rehydrate, mono_now()).await; + + tokio::spawn(flowspec_manager_task( + fs_manager, + fs_rx, + store.clone(), + )); + } } let selection = blackwall_flow::SelectionConfig { diff --git a/bin/blackwalld/src/shadow.rs b/bin/blackwalld/src/shadow.rs index cad3b1e..1c64db2 100644 --- a/bin/blackwalld/src/shadow.rs +++ b/bin/blackwalld/src/shadow.rs @@ -46,29 +46,43 @@ impl AuditShadowRecorder { #[async_trait::async_trait] impl ShadowRecorder for AuditShadowRecorder { async fn record(&self, action: ShadowAction) { - let (plane, verb, target, counter): (&str, &str, String, &AtomicU64) = match &action { + // `target` is a short display string for the INFO log only; `detail` + // is the structured JSON persisted to `audit_log` so `/v1/audit` + // consumers read fields (prefix/next_hop/dst/proto/…) rather than + // regex over a Debug blob. + let (plane, verb, target, detail, counter): ( + &str, + &str, + String, + serde_json::Value, + &AtomicU64, + ) = match &action { ShadowAction::RtbhAnnounce(r) => ( "rtbh", "announce", - format!("{r:?}"), + r.prefix.to_string(), + route_detail("rtbh", "announce", r), &self.metrics.rtbh_announce, ), ShadowAction::RtbhWithdraw(p) => ( "rtbh", "withdraw", p.to_string(), + serde_json::json!({ "plane": "rtbh", "verb": "withdraw", "prefix": p.to_string() }), &self.metrics.rtbh_withdraw, ), ShadowAction::FlowSpecAnnounce(r) => ( "flowspec", "announce", - format!("{r:?}"), + r.dst.to_string(), + flowspec_detail("flowspec", "announce", r), &self.metrics.flowspec_announce, ), ShadowAction::FlowSpecWithdraw(r) => ( "flowspec", "withdraw", - format!("{r:?}"), + r.dst.to_string(), + flowspec_detail("flowspec", "withdraw", r), &self.metrics.flowspec_withdraw, ), }; @@ -79,7 +93,6 @@ impl ShadowRecorder for AuditShadowRecorder { target = %target, "shadow: would mitigate (logged, not applied)" ); - let detail = serde_json::json!({ "plane": plane, "verb": verb, "target": target }); if let Err(err) = self .store .record_audit("shadow", &format!("shadow.{plane}.{verb}"), &detail) @@ -90,6 +103,40 @@ impl ShadowRecorder for AuditShadowRecorder { } } +/// Structured audit detail for an RTBH route: its prefix, next hop, and +/// communities (as `asn:value` strings) as their own JSON fields, so audit +/// consumers read fields rather than a Debug blob. +fn route_detail(plane: &str, verb: &str, r: &blackwall_bgp::Route) -> serde_json::Value { + let communities: Vec = r + .communities + .iter() + .map(|(asn, value)| format!("{asn}:{value}")) + .collect(); + serde_json::json!({ + "plane": plane, + "verb": verb, + "prefix": r.prefix.to_string(), + "next_hop": r.next_hop.to_string(), + "communities": communities, + }) +} + +/// Structured audit detail for a FlowSpec rule: destination, protocol, +/// destination port, and rate (bytes/sec) as their own JSON fields. +fn flowspec_detail(plane: &str, verb: &str, r: &blackwall_bgp::FlowSpecRule) -> serde_json::Value { + let rate = match &r.action { + blackwall_bgp::FlowAction::TrafficRate(rate) => *rate, + }; + serde_json::json!({ + "plane": plane, + "verb": verb, + "dst": r.dst.to_string(), + "protocol": r.protocol, + "dst_port": r.dst_port, + "rate": rate, + }) +} + /// [`XdpExecutor`] used in place of the live eBPF map writer when the /// `shadow` config directive is set: install actions (`Block`/`RateLimit`) /// are logged, metered into [`ShadowMetrics`], and audit-logged instead of