feat(shadow): global shadow mode — log/record mitigations, act on nothing

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<Arc<std::sync::Mutex<Vec<blackwall_flow::AgentStat>>>>,
+    /// Shadow-mode "would mitigate" counters (RTBH/FlowSpec/XDP); `None`
+    /// outside the flow daemon (no RTBH/FlowSpec/XDP managers to shadow).
+    pub shadow: Option<Arc<crate::shadow::ShadowMetrics>>,
 }
 
 /// 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<String> {
     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<String> {
+    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()

bin/blackwalld/src/shadow.rs

@@ -0,0 +1,252 @@
+//! Concrete shadow recorder: logs, meters, and audit-logs intended mitigations.
+//! 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;
+
+/// 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<blackwall_state::Store>,
+    metrics: Arc<ShadowMetrics>,
+}
+
+impl AuditShadowRecorder {
+    /// Build a recorder that audits to `store` and meters into `metrics`.
+    pub fn new(store: Arc<blackwall_state::Store>, metrics: Arc<ShadowMetrics>) -> Self {
+        Self { store, metrics }
+    }
+}
+
+#[async_trait::async_trait]
+impl ShadowRecorder for AuditShadowRecorder {
+    async fn record(&self, action: ShadowAction) {
+        // `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",
+                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",
+                r.dst.to_string(),
+                flowspec_detail("flowspec", "announce", r),
+                &self.metrics.flowspec_announce,
+            ),
+            ShadowAction::FlowSpecWithdraw(r) => (
+                "flowspec",
+                "withdraw",
+                r.dst.to_string(),
+                flowspec_detail("flowspec", "withdraw", r),
+                &self.metrics.flowspec_withdraw,
+            ),
+        };
+        counter.fetch_add(1, Ordering::Relaxed);
+        tracing::info!(
+            plane,
+            verb,
+            target = %target,
+            "shadow: would mitigate (logged, not applied)"
+        );
+        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)");
+        }
+    }
+}
+
+/// 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<String> = 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
+/// 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<blackwall_state::Store>,
+    metrics: Arc<ShadowMetrics>,
+}
+
+impl ShadowXdpExecutor {
+    /// Build an executor that audits to `store` and meters into `metrics`.
+    pub fn new(store: Arc<blackwall_state::Store>, metrics: Arc<ShadowMetrics>) -> 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<blackwall_xdp::XdpDataplane>),
+    /// 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,
+        }
+    }
+}

crates/blackwall-config/src/parser.rs

@@ -28,6 +28,7 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
     let mut xdp: Option<XdpConfig> = None;
     let mut stateless_tcp_ports: Vec<u16> = Vec::new();
     let mut pops: Vec<PopEntry> = Vec::new();
+    let mut shadow = false;
 
     let mut i = 0;
     while i < lines.len() {
@@ -772,6 +773,16 @@ pub fn parse(lines: &[Line]) -> Result<Policy, ConfigError> {
                     }
                 }
             }
+            "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<Policy, ConfigError> {
         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(

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<u16>,
+    /// Shadow mode: log + record + meter mitigations (RTBH/FlowSpec/XDP)
+    /// without applying them. `false` (the default) is live.
+    pub shadow: bool,
 }

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());