C2b-1: FlowSpec auto-mitigation core

.github/workflows/ci.yml

@@ -136,6 +136,14 @@ jobs:
           sudo timeout --kill-after=15s 300s env "PATH=$HOME/.cargo/bin:$PATH" "HOME=$HOME" \
             "CARGO_HOME=${CARGO_HOME:-$HOME/.cargo}" "RUSTUP_HOME=${RUSTUP_HOME:-$HOME/.rustup}" \
             ./target/debug/lab test crates/blackwall-lab/scenarios/flowspec-bird.kdl lab-junit-flowspec.xml
+      - name: Run the lab gate (flowspec-auto)
+        timeout-minutes: 6
+        run: |
+          trap 'rc=$?; exec 1>&3 2>&4; cat lab-gate.log; exit $rc' EXIT
+          exec 3>&1 4>&2 >lab-gate.log 2>&1
+          sudo timeout --kill-after=15s 300s env "PATH=$HOME/.cargo/bin:$PATH" "HOME=$HOME" \
+            "CARGO_HOME=${CARGO_HOME:-$HOME/.cargo}" "RUSTUP_HOME=${RUSTUP_HOME:-$HOME/.rustup}" \
+            ./target/debug/lab test crates/blackwall-lab/scenarios/flowspec-auto-bird.kdl lab-junit-flowspec-auto.xml
       - name: Upload JUnit
         if: always()
         uses: actions/upload-artifact@v4
@@ -152,3 +160,4 @@ jobs:
             lab-junit-deception-resilience.xml
             lab-junit-rtbh.xml
             lab-junit-flowspec.xml
+            lab-junit-flowspec-auto.xml

CHANGELOG.md

@@ -15,6 +15,7 @@ All notable changes to this project are documented here, following
 - nftables rendering bound the `prerouting` filter chain to the managed interface (`type filter hook prerouting … device`), which the kernel rejects — only ingress/egress chains may bind a device — so `blackwall-nft::apply` failed on any real ruleset. The chain is now unbound and classification is scoped per-rule with an `iifname` match, the correct pattern for a prerouting filter chain. Because an unbound chain runs for every interface, the closed posture (`default_state` closed) is now enforced by an explicit interface-scoped terminal `drop` rule instead of a chain-wide drop policy, so it no longer black-holes loopback or other-interface host traffic. Caught by the new deception↔scanner lab gate, the first end-to-end run of `apply` against a real `nft`.
 
 ### Added
+- FlowSpec auto-mitigation core (sub-project C, C2b-1): a detected attack now automatically becomes the *right* mitigation. A pure concentration-based selection inspects the detection's fingerprint (`proto` + `top_ports`): when a small set of flows carries most of the attack it emits FlowSpec drop rules (*drop UDP dport 53 → victim/32*), leaving the victim's other services up; when the attack is diffuse across many ports it falls back to RTBH's whole-IP blackhole (which FlowSpec can't scope). A `FlowSpecController` + single-owner `FlowSpecManager` mirror the RTBH controller/manager (eligibility, concurrent-rule cap, hold-down + deferred-withdraw + TTL, origin, mirror self-heal) and share the one iBGP session via an extended `BgpExecutor`; a `SelectorSink` routes each detection to the FlowSpec or RTBH manager. A new `flowspec-auto-bird` lab gate proves both paths end to end against real **BIRD2**: a concentrated detection installs the flow rule in `flow4tab`, and a diffuse one installs the `/32` blackhole (community `65535:666`). Persistence, the `flowspec` config directive, the `blackwalld flowspec` operator CLI, and the daemon wiring are C2b-2 follow-ons.
 - FlowSpec codec (sub-project C, C2a): finer-grained BGP mitigation than RTBH — drop (or rate-limit) *just the attack flow* instead of null-routing the whole victim IP. A new byte-exact `blackwall-bgp` FlowSpec codec encodes rules for both families (RFC 8955 v4 + RFC 8956 v6, SAFI 133): the minimal DDoS match set — destination-prefix, IP-protocol, destination-port (with the RFC 8955 numeric operator-value encoding) — carrying a traffic-rate action (rate `0.0` = discard, `N` = rate-limit). The OPEN now advertises the FlowSpec capabilities, and the iBGP session gains a FlowSpec inject path (announce/withdraw, re-announced on reconnect like unicast routes). A new `flowspec-bird` lab gate proves an injected rule (`drop UDP dport 53 → 203.0.113.7/32`) reaches real **BIRD2** and validates + installs into its `flow4` table with the full match (`dst 203.0.113.7/32; proto 17; dport 53`). The controller/policy layer, detector auto-trigger, operator CLI, persistence, and richer match components/actions are C2b follow-ons.
 - RTBH control plane (sub-project C, C1c): makes RTBH durable, production-wired, and operable. Active blackholes are persisted to PostgreSQL (`rtbh_blackholes`) and **re-announced on daemon restart** (a crash never drops protection); `blackwalld flow` now auto-blackholes detected attacks through the live C1a BGP speaker, so the D→C link runs in production, not just the lab. A new `blackwalld rtbh add/remove/list` operator CLI records intent to an append-only `rtbh_requests` log (with `created_by` for attribution) that the daemon drains and applies. Internally, a single-owner `RtbhManager` task owns the controller + BGP session and is the sole reconciler (no shared-lock race; removals are explicit events, never inferred from DB absence), superseding the C1b `RtbhSink`. RTBH peering + policy is configured via a new `rtbh` config directive (BGP peer/ASN/router-id, blackhole community, per-family next-hops, cap, hold-down, TTL). `blackwall-rtbh` stays free of a DB dependency via `BgpExecutor`/`BlackholeJournal` traits it defines and `blackwall-bgp`/`blackwall-state` implement. The `rtbh-bird` lab gate now proves both an auto-detected and an operator-manual `/32` (community `65535:666`) reach real BIRD.
 - RTBH (sub-project C, C1b): a `blackwall-rtbh` crate turning detected attacks into BGP blackholes. A pure `RtbhController` maps detection events (and manual triggers) to `/32`/`/128` blackhole announce/withdraw decisions with eligibility (own-prefix only), a concurrent-blackhole cap, and hold-down anti-flap, executed against the C1a BGP speaker. A new `rtbh-bird` lab gate proves a blackhole `/32` (community `65535:666`) reaches real BIRD — the first detection→mitigation link.

README.md

@@ -10,10 +10,12 @@ the real thing. Blackwall is built for high packet rates and multi-tenant hostin
 nftables data plane today and an XDP/eBPF fast path, BGP scrubbing, and DNS fast-flux on the
 roadmap.
 
-> ⚠️ **Status:** early development. Milestone 1 (the core foundation) is complete; the deception
-> engine and later layers are in progress — see [Roadmap](#roadmap). The current nftables ruleset
-> classifies policy structure but does **not** yet enforce deception/forwarding (that lands in
-> Milestone 2).
+> ⚠️ **Status:** active development. Sub-project **A** (deception firewall) is feature-complete
+> through M3 — the nftables data plane enforces deception (TPROXY/NFQUEUE → honeypot engine) and
+> real-service forwarding, with protocol emulators, Incus discovery, CAKE shaping, and DNS/banner
+> fast-flux. Sub-project **D** ships volumetric detection (D1), and **C** (BGP control plane) ships
+> RTBH end to end plus FlowSpec auto-mitigation. The remaining layers — A·M4 (API/ops) and B (the
+> XDP/eBPF DDoS data plane) — are not yet built. See the [Roadmap](#roadmap).
 
 ## How it works
 
@@ -151,6 +153,7 @@ Current scenarios (each a CI gate):
 | `deception-resilience` | A connection flood past the deception engine's `max_concurrent` cap proves its DDoS-defense is correct — drop-at-cap is enforced, legit deception still gets `SSH-2.0`, and the engine survives. A **resilience/correctness** gate, not a throughput benchmark (realistic-scale stress needs kernel-bypass, tracked separately). |
 | `rtbh-bird` | The `RtbhManager` announces both an auto-detected and an operator-manual `/32` blackhole (community `65535:666`, RFC 7999) via the native BGP speaker; real **BIRD2** must show both routes carrying that community — the detection→mitigation (D→C) loop, auto and manual, end to end. |
 | `flowspec-bird` | The native speaker injects a BGP **FlowSpec** rule (RFC 8955, SAFI 133) — *drop UDP dport 53 → 203.0.113.7/32* — over iBGP; real **BIRD2** must validate and install it into its `flow4` table with the full match (`dst 203.0.113.7/32; proto 17; dport 53`) — finer-grained mitigation than a whole-IP blackhole. |
+| `flowspec-auto-bird` | The concentration-based selector routes a synthetic detection to the right mitigation: a *concentrated* attack (one dominant port) auto-installs a FlowSpec drop rule in real **BIRD2**'s `flow4tab`, while a *diffuse* attack auto-installs an RTBH `/32` blackhole — the auto-mitigation decision (`FlowSpec` vs `RTBH`) proven end to end. |
 
 The architecture is pure-core / thin-IO: the topology compiler, address allocator, config
 renderers, and report serializers are unit-tested to the 90% gate; the netns/process executor is
@@ -180,19 +183,36 @@ generation to exercise the XDP/eBPF data plane.
 
 ## Roadmap
 
-Blackwall is built as four independent sub-projects, each delivered in milestones:
+Blackwall is built as four independent sub-projects, each delivered in milestones.
 
-**A — Deception firewall + orchestrator** *(in progress)*
+**A — Deception firewall + orchestrator**
 - ✅ **M1 — Core foundation:** workspace, domain/policy model, config DSL, PostgreSQL state,
   nftables render + atomic apply, CLI.
-- ⏳ **M2 — Deception engine:** two-tier stateless + interactive honeypot with protocol emulators.
-- ⏳ **M3 — Discovery, shaping, flux:** host/Incus discovery, automatic CAKE shaping + speedtests,
-  signature rotation, DNS fast-flux (Knot DNS).
+- ✅ **M2 — Deception engine:** two-tier stateless + interactive honeypot; SSH/HTTP/SMTP/Redis/
+  MySQL/PostgreSQL emulators; TPROXY/NFQUEUE transports; connection cap + session timeout.
+- ✅ **M3 — Discovery, shaping, flux:** host/Incus discovery + reconciler, automatic CAKE shaping
+  from multi-provider speedtests, banner fast-flux, DNS fast-flux (RFC 2136 + TSIG to Knot).
 - ⏳ **M4 — API & ops:** tenant-scoped REST API, Prometheus metrics, daemon supervision.
 
-**B — DDoS data plane** — XDP/eBPF + AF_XDP fast drop, SYNPROXY, conntrack, rate limiting.
-**C — ISP/BGP control plane** — own-ASN prefix announcement, RTBH, FlowSpec, scrubbing, dn42.
-**D — Detection & telemetry** — sFlow/NetFlow/IPFIX ingest driving B and C.
+**B — DDoS data plane** *(not started)* — XDP/eBPF + AF_XDP fast drop, SYNPROXY, conntrack, rate
+limiting. The DDoS-lab traffic-generation foundation (`blackwall-trafficgen`) is in place ahead of it.
+
+**C — ISP/BGP control plane** *(in progress)* — a native injection-only iBGP speaker to BIRD.
+- ✅ **C1 — RTBH:** byte-exact BGP codec + speaker (C1a); pure blackhole controller (C1b);
+  the full control plane — Postgres persistence, detector auto-trigger, operator CLI (C1c).
+- ✅ **C2a — FlowSpec codec:** RFC 8955/8956 (SAFI 133) rule encoding + speaker inject path.
+- ✅ **C2b-1 — Auto-mitigation core:** concentration-based selection routes a detection to a
+  flow-scoped FlowSpec drop (concentrated) or an RTBH blackhole (diffuse).
+- ⏳ **C2b-2 — FlowSpec control plane:** persistence, `flowspec` config, operator CLI, daemon wiring.
+- ⏳ **C3 — Looking-glass, C4 — auto-peering, scrubbing, dn42** *(later).*
+
+**D — Detection & telemetry** *(in progress)*
+- ✅ **D1 — Volumetric detection:** sFlow v5 ingest + sliding-window threshold detector driving C.
+- ⏳ **D2+ — NetFlow/IPFIX, adaptive baseline detection** *(later).*
+
+**Integration lab** *(cross-cutting)* — a reproducible netns harness (`blackwall-lab`) with one
+`lab` command that runs identically locally and in CI. Every component is gated against the real
+peer software (BIRD, Knot, CAKE, hsflowd, nftables) — see [Integration lab](#integration-lab-blackwall-lab).
 
 ## License
 

crates/blackwall-flow/src/lib.rs

@@ -5,12 +5,16 @@ mod collector_net;
 mod detector;
 mod error;
 mod observation;
+mod select;
 mod sflow;
 mod sink;
 
 pub use collector_net::run_collector;
 pub use detector::{AttackKind, Detection, DetectionEvent, Detector, Severity, ThresholdDetector};
 pub use error::FlowError;
 pub use observation::FlowObservation;
+pub use select::{select, FlowRule, Mitigation, SelectionConfig};
 pub use sflow::decode_datagram;
-pub use sink::{ChannelSink, FanoutSink, LogSink, MitigationSink};
+pub use sink::{
+    ChannelSink, FanoutSink, FlowMitigationEvent, LogSink, MitigationSink, SelectorSink,
+};

crates/blackwall-flow/src/select.rs

@@ -0,0 +1,206 @@
+//! Pure concentration-based mitigation selection: a flow-concentrated attack
+//! becomes FlowSpec drop rules; a diffuse attack falls back to RTBH.
+
+use crate::detector::Detection;
+use std::net::IpAddr;
+
+/// A neutral flow-scoped rule (no BGP types — keeps `blackwall-flow` BGP-free).
+#[derive(Debug, Clone, PartialEq)]
+pub struct FlowRule {
+    /// The victim address (a host route is derived downstream).
+    pub dst: IpAddr,
+    /// IP protocol (e.g. 17 = UDP).
+    pub proto: u8,
+    /// Destination port.
+    pub dst_port: u16,
+    /// Traffic-rate action in bytes/sec; `0.0` = drop.
+    pub rate: f32,
+}
+
+/// The chosen mitigation for a detection.
+#[derive(Debug, Clone, PartialEq)]
+pub enum Mitigation {
+    /// Drop the listed flows (leaving the victim's other services up).
+    FlowSpec(Vec<FlowRule>),
+    /// Blackhole the whole victim IP (diffuse attack; FlowSpec can't scope it).
+    Rtbh,
+}
+
+/// Tunables for [`select`].
+#[derive(Debug, Clone)]
+pub struct SelectionConfig {
+    /// Cumulative top-port weight fraction that counts as "concentrated".
+    pub concentration: f64,
+    /// Max distinct flows a FlowSpec mitigation may emit.
+    pub max_flows: usize,
+    /// Traffic-rate for emitted rules (bytes/sec; `0.0` = drop).
+    pub rate: f32,
+}
+
+/// Choose FlowSpec (flow-scoped drop) vs RTBH (whole-IP) for a detection.
+///
+/// FlowSpec is chosen only when a small set (≤ `max_flows`) of the top
+/// destination ports carries at least `concentration` of the flow — i.e. when
+/// dropping those flows actually stops the attack without taking the victim
+/// fully offline. Otherwise (diffuse, no ports, or no protocol) falls back to RTBH.
+#[must_use]
+pub fn select(d: &Detection, cfg: &SelectionConfig) -> Mitigation {
+    if d.proto == 0 || d.top_ports.is_empty() {
+        return Mitigation::Rtbh;
+    }
+    let mut ports = d.top_ports.clone();
+    ports.sort_by(|a, b| b.1.total_cmp(&a.1)); // weight desc
+    let mut cumulative = 0.0_f64;
+    let mut chosen: Vec<u16> = Vec::new();
+    for (port, weight) in ports {
+        if chosen.len() >= cfg.max_flows {
+            break;
+        }
+        chosen.push(port);
+        cumulative += weight;
+        if cumulative >= cfg.concentration {
+            return Mitigation::FlowSpec(
+                chosen
+                    .into_iter()
+                    .map(|p| FlowRule {
+                        dst: d.target,
+                        proto: d.proto,
+                        dst_port: p,
+                        rate: cfg.rate,
+                    })
+                    .collect(),
+            );
+        }
+    }
+    Mitigation::Rtbh
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::detector::{AttackKind, Detection, Severity};
+
+    fn det(proto: u8, ports: Vec<(u16, f64)>) -> Detection {
+        Detection {
+            target: "203.0.113.7".parse().unwrap(),
+            kind: AttackKind::Volumetric,
+            observed_pps: 1.0,
+            observed_bps: 1.0,
+            proto,
+            top_sources: vec![],
+            top_ports: ports,
+            severity: Severity::High,
+            first_seen_ms: 0,
+            last_seen_ms: 0,
+        }
+    }
+    fn cfg() -> SelectionConfig {
+        SelectionConfig {
+            concentration: 0.8,
+            max_flows: 4,
+            rate: 0.0,
+        }
+    }
+
+    #[test]
+    fn concentrated_single_port_selects_flowspec() {
+        let m = select(&det(17, vec![(53, 0.95)]), &cfg());
+        assert_eq!(
+            m,
+            Mitigation::FlowSpec(vec![FlowRule {
+                dst: "203.0.113.7".parse().unwrap(),
+                proto: 17,
+                dst_port: 53,
+                rate: 0.0
+            }])
+        );
+    }
+
+    #[test]
+    fn concentrated_multi_port_under_cap_selects_flowspec() {
+        // two ports together cross 0.8; both become rules.
+        let m = select(&det(6, vec![(80, 0.5), (443, 0.4), (22, 0.1)]), &cfg());
+        let Mitigation::FlowSpec(rules) = m else {
+            panic!("expected FlowSpec")
+        };
+        assert_eq!(rules.len(), 2);
+        assert_eq!(rules[0].dst_port, 80);
+        assert_eq!(rules[1].dst_port, 443);
+    }
+
+    #[test]
+    fn diffuse_falls_back_to_rtbh() {
+        // Genuinely diffuse: the top 4 ports (max_flows) sum to only 0.4, well
+        // short of 0.8 — no small flow set can stop the attack, so RTBH.
+        let m = select(
+            &det(
+                17,
+                vec![
+                    (1, 0.1),
+                    (2, 0.1),
+                    (3, 0.1),
+                    (4, 0.1),
+                    (5, 0.1),
+                    (6, 0.1),
+                    (7, 0.1),
+                    (8, 0.1),
+                ],
+            ),
+            &cfg(),
+        );
+        assert_eq!(m, Mitigation::Rtbh);
+    }
+
+    #[test]
+    fn single_port_100pct_with_max_flows_1_selects_flowspec() {
+        // Textbook FlowSpec case: 100% on one port, tightest budget (max_flows=1)
+        // — must be a surgical flow rule, NOT a whole-IP blackhole.
+        let c = SelectionConfig {
+            concentration: 0.8,
+            max_flows: 1,
+            rate: 0.0,
+        };
+        let m = select(&det(17, vec![(53, 1.0)]), &c);
+        let Mitigation::FlowSpec(rules) = m else {
+            panic!("expected FlowSpec, got {m:?}")
+        };
+        assert_eq!(rules.len(), 1);
+        assert_eq!(rules[0].dst_port, 53);
+    }
+
+    #[test]
+    fn ranks_ports_by_weight_before_selecting() {
+        // Unsorted input: the dominant port (443, 0.9) is listed last; it must
+        // still be the chosen flow.
+        let m = select(&det(6, vec![(22, 0.05), (80, 0.05), (443, 0.9)]), &cfg());
+        let Mitigation::FlowSpec(rules) = m else {
+            panic!("expected FlowSpec")
+        };
+        assert_eq!(rules.len(), 1);
+        assert_eq!(rules[0].dst_port, 443);
+    }
+
+    #[test]
+    fn empty_ports_or_no_proto_is_rtbh() {
+        assert_eq!(select(&det(17, vec![]), &cfg()), Mitigation::Rtbh);
+        assert_eq!(select(&det(0, vec![(53, 0.99)]), &cfg()), Mitigation::Rtbh);
+    }
+
+    #[test]
+    fn respects_max_flows_cap() {
+        // 5 ports each 0.19 reach 0.8 only at the 5th, but max_flows=4 -> diffuse.
+        let c = SelectionConfig {
+            concentration: 0.8,
+            max_flows: 4,
+            rate: 0.0,
+        };
+        let m = select(
+            &det(
+                17,
+                vec![(1, 0.19), (2, 0.19), (3, 0.19), (4, 0.19), (5, 0.19)],
+            ),
+            &c,
+        );
+        assert_eq!(m, Mitigation::Rtbh);
+    }
+}