diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5ba2d14..5c75b46 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ada2f1..7eeca6e 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 6ec8115..ebe3151 100644 --- a/README.md +++ b/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 diff --git a/crates/blackwall-flow/src/lib.rs b/crates/blackwall-flow/src/lib.rs index 2031a4b..b5ac502 100644 --- a/crates/blackwall-flow/src/lib.rs +++ b/crates/blackwall-flow/src/lib.rs @@ -5,6 +5,7 @@ mod collector_net; mod detector; mod error; mod observation; +mod select; mod sflow; mod sink; @@ -12,5 +13,8 @@ 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, +}; diff --git a/crates/blackwall-flow/src/select.rs b/crates/blackwall-flow/src/select.rs new file mode 100644 index 0000000..d1f697b --- /dev/null +++ b/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), + /// 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 = 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); + } +} diff --git a/crates/blackwall-flow/src/sink.rs b/crates/blackwall-flow/src/sink.rs index 9e700ee..ffd7ed9 100644 --- a/crates/blackwall-flow/src/sink.rs +++ b/crates/blackwall-flow/src/sink.rs @@ -2,7 +2,9 @@ //! later B/C mitigation) plugs in here. use crate::detector::DetectionEvent; +use crate::select::{select, FlowRule, Mitigation, SelectionConfig}; use async_trait::async_trait; +use std::net::IpAddr; use std::sync::Arc; use tokio::sync::mpsc; @@ -72,6 +74,98 @@ impl MitigationSink for ChannelSink { } } +/// A mitigation instruction produced by [`SelectorSink`] for the FlowSpec side. +#[derive(Debug, Clone)] +pub enum FlowMitigationEvent { + /// Install FlowSpec drop rules for `target`. + Open { + /// The victim address. + target: IpAddr, + /// The flow-scoped rules to install. + rules: Vec, + }, + /// The attack against `target` is still ongoing; refresh/keep-alive it. + Update { + /// The victim address. + target: IpAddr, + }, + /// The attack against `target` has cleared; withdraw its rules. + Clear { + /// The victim address. + target: IpAddr, + }, +} + +/// A sink that routes each detection to the FlowSpec or RTBH manager, using +/// [`select`] to pick the mitigation for newly-opened attacks. +/// +/// `Opened` detections are classified via `select` and routed to exactly one +/// of `flowspec_tx` (FlowSpec) or `rtbh_tx` (RTBH). `Updated` and `Cleared` +/// events are broadcast to both, since either manager may be holding state +/// for the target. Forwarding is best-effort via `try_send`: a full or closed +/// channel is logged with `tracing::warn!` and otherwise ignored; `handle` +/// never panics and never blocks. +pub struct SelectorSink { + flowspec_tx: mpsc::Sender, + rtbh_tx: mpsc::Sender, + cfg: SelectionConfig, +} + +impl SelectorSink { + /// Build a `SelectorSink` that routes via `select(_, &cfg)`, forwarding + /// FlowSpec mitigations into `flowspec_tx` and RTBH mitigations (plus all + /// `Updated`/`Cleared` events) into `rtbh_tx`. + pub fn new( + flowspec_tx: mpsc::Sender, + rtbh_tx: mpsc::Sender, + cfg: SelectionConfig, + ) -> Self { + Self { + flowspec_tx, + rtbh_tx, + cfg, + } + } + + fn send_flowspec(&self, event: FlowMitigationEvent) { + if let Err(err) = self.flowspec_tx.try_send(event) { + tracing::warn!(%err, "dropping flowspec mitigation event: channel unavailable"); + } + } + + fn send_rtbh(&self, event: DetectionEvent) { + if let Err(err) = self.rtbh_tx.try_send(event) { + tracing::warn!(%err, "dropping rtbh detection event: channel unavailable"); + } + } +} + +#[async_trait] +impl MitigationSink for SelectorSink { + async fn handle(&self, event: &DetectionEvent) { + match event { + DetectionEvent::Opened(d) => match select(d, &self.cfg) { + Mitigation::FlowSpec(rules) => self.send_flowspec(FlowMitigationEvent::Open { + target: d.target, + rules, + }), + Mitigation::Rtbh => self.send_rtbh(DetectionEvent::Opened(d.clone())), + }, + DetectionEvent::Updated(d) => { + self.send_flowspec(FlowMitigationEvent::Update { target: d.target }); + self.send_rtbh(DetectionEvent::Updated(d.clone())); + } + DetectionEvent::Cleared { target, at_ms } => { + self.send_flowspec(FlowMitigationEvent::Clear { target: *target }); + self.send_rtbh(DetectionEvent::Cleared { + target: *target, + at_ms: *at_ms, + }); + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -143,4 +237,111 @@ mod tests { }) .await; // must not panic } + + use crate::detector::{AttackKind, Detection, Severity}; + + fn det_base(proto: u8, top_ports: Vec<(u16, f64)>) -> Detection { + Detection { + target: IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), + kind: AttackKind::Volumetric, + observed_pps: 1_000.0, + observed_bps: 8_000.0, + proto, + top_sources: vec![], + top_ports, + severity: Severity::High, + first_seen_ms: 0, + last_seen_ms: 0, + } + } + + /// A concentrated attack: a single dominant port well past the default + /// concentration threshold, so `select` chooses FlowSpec. + fn det_concentrated() -> Detection { + det_base(17, vec![(53, 0.95)]) + } + + /// A diffuse attack: weight spread thin across many ports, so no small + /// flow set clears the concentration threshold and `select` chooses RTBH. + fn det_diffuse() -> Detection { + det_base( + 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), + ], + ) + } + + fn selector_cfg() -> SelectionConfig { + SelectionConfig { + concentration: 0.8, + max_flows: 4, + rate: 0.0, + } + } + + #[tokio::test] + async fn concentrated_routes_to_flowspec_only() { + let (ftx, mut frx) = mpsc::channel(8); + let (rtx, mut rrx) = mpsc::channel(8); + let s = SelectorSink::new(ftx, rtx, selector_cfg()); + s.handle(&DetectionEvent::Opened(det_concentrated())).await; + assert!(matches!( + frx.try_recv(), + Ok(FlowMitigationEvent::Open { .. }) + )); + assert!(rrx.try_recv().is_err()); // RTBH not routed + } + + #[tokio::test] + async fn diffuse_routes_to_rtbh_only() { + let (ftx, mut frx) = mpsc::channel(8); + let (rtx, mut rrx) = mpsc::channel(8); + let s = SelectorSink::new(ftx, rtx, selector_cfg()); + s.handle(&DetectionEvent::Opened(det_diffuse())).await; + assert!(matches!(rrx.try_recv(), Ok(DetectionEvent::Opened(_)))); + assert!(frx.try_recv().is_err()); // FlowSpec not routed + } + + #[tokio::test] + async fn clear_broadcasts_to_both() { + let (ftx, mut frx) = mpsc::channel(8); + let (rtx, mut rrx) = mpsc::channel(8); + let s = SelectorSink::new(ftx, rtx, selector_cfg()); + let target = IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)); + s.handle(&DetectionEvent::Cleared { target, at_ms: 42 }) + .await; + assert!(matches!( + frx.try_recv(), + Ok(FlowMitigationEvent::Clear { target: t }) if t == target + )); + assert!(matches!( + rrx.try_recv(), + Ok(DetectionEvent::Cleared { target: t, at_ms: 42 }) if t == target + )); + } + + #[tokio::test] + async fn closed_channel_does_not_panic() { + let (ftx, frx) = mpsc::channel(8); + let (rtx, rrx) = mpsc::channel(8); + drop(frx); + drop(rrx); + let s = SelectorSink::new(ftx, rtx, selector_cfg()); + s.handle(&DetectionEvent::Opened(det_concentrated())).await; + s.handle(&DetectionEvent::Opened(det_diffuse())).await; + s.handle(&DetectionEvent::Updated(det_concentrated())).await; + s.handle(&DetectionEvent::Cleared { + target: IpAddr::V4(Ipv4Addr::new(203, 0, 113, 7)), + at_ms: 0, + }) + .await; // must not panic + } } diff --git a/crates/blackwall-lab/scenarios/flowspec-auto-bird.kdl b/crates/blackwall-lab/scenarios/flowspec-auto-bird.kdl new file mode 100644 index 0000000..86e4259 --- /dev/null +++ b/crates/blackwall-lab/scenarios/flowspec-auto-bird.kdl @@ -0,0 +1,36 @@ +// FlowSpec-auto gate: the pure concentration-based selector (`select()`) picks +// FlowSpec vs RTBH per detection, and the chosen manager announces it over a +// real iBGP session to a real BIRD2 peer — proving the auto-mitigation path +// end to end: a concentrated detection becomes a FlowSpec drop rule; a +// diffuse detection becomes an RTBH /32 blackhole. Mirrors flowspec-bird.kdl +// and rtbh-bird.kdl, combined behind the selector. +// +// Validation gotcha (RFC 8955 §6, same as flowspec-bird.kdl): BIRD only +// accepts a FlowSpec rule whose destination is covered by a route in the +// unicast table from the same origin AS ("safe update" check). Resolved by +// having the interop driver announce a covering unicast route +// (203.0.113.0/24 via 10.0.0.1) over the same session before either +// mitigation. +topology "flowspec-auto-bird" { + node "peer" { + daemon "bird" local-as="214806" neighbor-node="speaker" neighbor-as="214806" import="all" passive="yes" flowspec="yes" + } + node "speaker" { + run "speaker" \ + cmd="target/debug/lab-tests/blackwall-rtbh-flowspec_auto_interop --ignored --nocapture" \ + env="BW_BGP_PEER={peer.addr}:179" + } + link "peer" "speaker" subnet="10.0.0.0/30" +} + +scenario "flowspec-auto-reaches-bird" { + step wait node="peer" until="bgp-established" timeout="20s" + // Concentrated detection (203.0.113.7, proto 17, dport 53 @ 95%) selects + // FlowSpec: BIRD must show the flow-scoped rule in flow4tab (note: `flow4` + // is a reserved BIRD keyword — the table is `flow4tab`). + step assert node="peer" cmd="birdc show route table flow4tab" contains="dst 203.0.113.7/32; proto 17; dport 53" timeout="15s" + // Diffuse detection (203.0.113.8, weight spread across 8 ports) selects + // RTBH: BIRD must show the /32 blackhole carrying the well-known + // blackhole community (65535,666). + step assert node="peer" cmd="birdc show route 203.0.113.8/32 all" contains="(65535,666)" timeout="15s" +} diff --git a/crates/blackwall-rtbh/src/controller.rs b/crates/blackwall-rtbh/src/controller.rs index 2fba275..7a1e26d 100644 --- a/crates/blackwall-rtbh/src/controller.rs +++ b/crates/blackwall-rtbh/src/controller.rs @@ -307,7 +307,7 @@ impl RtbhController { /// Construct a host prefix for a target address. /// /// Returns `/32` for IPv4 or `/128` for IPv6. -fn host_prefix(target: IpAddr) -> IpNet { +pub(crate) fn host_prefix(target: IpAddr) -> IpNet { match target { IpAddr::V4(a) => IpNet::V4(ipnet::Ipv4Net::new(a, 32).expect("v4 /32")), IpAddr::V6(a) => IpNet::V6(ipnet::Ipv6Net::new(a, 128).expect("v6 /128")), diff --git a/crates/blackwall-rtbh/src/flowspec_controller.rs b/crates/blackwall-rtbh/src/flowspec_controller.rs new file mode 100644 index 0000000..7043c82 --- /dev/null +++ b/crates/blackwall-rtbh/src/flowspec_controller.rs @@ -0,0 +1,614 @@ +//! The pure FlowSpec decision engine. No I/O; deterministic given an injected `now`. +//! +//! The FlowSpec analogue of [`crate::controller::RtbhController`]: instead of a single +//! blackhole per target, a target may have multiple active flow rules (one per +//! `(protocol, destination-port)` pair), each carrying its own traffic-rate action. +//! FlowSpec carries no next-hop, so unlike RTBH there is no `has_next_hop` guard. + +use crate::controller::{host_prefix, BlackholeOrigin}; +use blackwall_bgp::{FlowAction, FlowSpecRule}; +use ipnet::IpNet; +use std::collections::HashMap; +use std::net::IpAddr; +use std::time::Duration; + +/// Identity of one active FlowSpec rule: the target's host route, the matched +/// IP protocol, and the matched destination port. +pub type FlowKey = (IpNet, u8, u16); + +/// State for one active FlowSpec rule. +#[derive(Debug, Clone)] +struct ActiveEntry { + rule: FlowSpecRule, + announced_at: u64, + last_activity: u64, + origin: BlackholeOrigin, + clear_requested_at: Option, +} + +/// FlowSpec policy configuration. +/// +/// Defines the parameters for the FlowSpec decision engine: eligible prefixes, +/// a hard cap on concurrent rules, and anti-flap hold-down/TTL. FlowSpec carries +/// no next-hop (the traffic-rate action is per-rule), so there is no next-hop field. +#[derive(Debug, Clone)] +pub struct FlowSpecConfig { + /// Only targets inside these prefixes may have FlowSpec rules installed + /// (never foreign space). + pub eligible_prefixes: Vec, + /// Hard cap on concurrent active rules (summed across all targets). + pub max_rules: usize, + /// Minimum time a rule stays before a clear may withdraw it (anti-flap). + pub hold_down: Duration, + /// Maximum lifetime of an auto rule (hygiene backstop against a dropped + /// or missed clear); `None` disables the TTL. + pub max_ttl: Option, +} + +/// A decision the [`FlowSpecController`] emits for the sink to execute. +/// +/// Either announce a FlowSpec traffic-filter rule for a detected attack flow, +/// or withdraw a previously-announced rule. +#[derive(Debug, Clone, PartialEq)] +pub enum FlowSpecAction { + /// Announce a FlowSpec rule. + Announce(FlowSpecRule), + /// Withdraw a previously-announced FlowSpec rule. + Withdraw(FlowSpecRule), +} + +/// The pure FlowSpec decision engine. +/// +/// A stateful controller mapping `(target, rules)` installs to FlowSpec actions. +/// It enforces eligibility, a total-rule capacity cap, and hold-down anti-flap +/// logic — the FlowSpec analogue of [`crate::controller::RtbhController`], adapted +/// from a single blackholed IP to multiple `(protocol, port)` rules per target. +/// Pure-core: deterministic given injected `now` timestamps, no I/O. +#[derive(Debug)] +pub struct FlowSpecController { + config: FlowSpecConfig, + active: HashMap, +} + +impl FlowSpecController { + /// Create a controller with no active rules. + #[must_use] + pub fn new(config: FlowSpecConfig) -> Self { + Self { + config, + active: HashMap::new(), + } + } + + /// Install FlowSpec rules for a detected attack target. + /// + /// # Arguments + /// + /// * `target` - The attacked IP address; the host route (`/32`/`/128`) is derived + /// from it and used as the FlowSpec destination-prefix match for every rule. + /// * `rules` - `(protocol, destination-port, traffic-rate)` tuples to install. + /// * `now` - Current time in milliseconds since epoch (used for hold-down/TTL anchoring). + /// + /// # Returns + /// + /// An `Announce` action per newly-installed rule. A rule already active is + /// re-asserted (its pending clear, if any, is cancelled and its TTL anchor + /// refreshed) without emitting a new action. Ineligible targets or a full + /// capacity cap are ignored. + pub fn install( + &mut self, + target: IpAddr, + rules: &[(u8, u16, f32)], + now: u64, + ) -> Vec { + let host = host_prefix(target); + rules + .iter() + .flat_map(|&(protocol, dst_port, rate)| { + let rule = FlowSpecRule { + dst: host, + protocol: Some(protocol), + dst_port: Some(dst_port), + action: FlowAction::TrafficRate(rate), + }; + self.insert_rule(target, rule, now, BlackholeOrigin::Auto) + }) + .collect() + } + + /// Request the clear of every active rule for `target`. + /// + /// Applies the same anti-flap logic as [`crate::controller::RtbhController::on_event`]'s + /// `Cleared` handling, per rule: a `Manual` rule is never auto-cleared; a rule + /// still within its hold-down has the clear deferred (to a later [`Self::tick`]); + /// a rule at/after hold-down is withdrawn immediately. + /// + /// # Arguments + /// + /// * `target` - The IP address whose flows should be cleared. + /// * `now` - Current time in milliseconds since epoch. + /// + /// # Returns + /// + /// A `Withdraw` action for every rule of `target` that clears immediately. + pub fn clear_target(&mut self, target: IpAddr, now: u64) -> Vec { + let host = host_prefix(target); + let hold_ms = u64::try_from(self.config.hold_down.as_millis()).unwrap_or(u64::MAX); + let keys: Vec = self + .active + .keys() + .filter(|k| k.0 == host) + .copied() + .collect(); + let mut actions = Vec::new(); + for key in keys { + match self.active.get_mut(&key) { + // Manual rules are never auto-cleared. + Some(e) if matches!(e.origin, BlackholeOrigin::Manual) => {} + Some(e) if now.saturating_sub(e.announced_at) >= hold_ms => { + if let Some(e) = self.active.remove(&key) { + actions.push(FlowSpecAction::Withdraw(e.rule)); + } + } + Some(e) => { + e.clear_requested_at = Some(now); + } + None => {} + } + } + actions + } + + /// Manually install (or upgrade) a FlowSpec rule (for the operator CLI + the lab). + /// + /// If a rule with the same `(dst, protocol, dst_port)` is already active as + /// `Auto`, this upgrades it to `Manual` (and cancels any pending deferred + /// clear) instead of re-announcing. + /// + /// # Arguments + /// + /// * `rule` - The FlowSpec rule to install; `rule.dst` should be an eligible + /// host route. + /// * `now` - Current time in milliseconds since epoch. + /// + /// # Returns + /// + /// An `Announce` action if newly installed, empty vector if upgraded, + /// ineligible, or at cap. + pub fn manual_add(&mut self, rule: FlowSpecRule, now: u64) -> Vec { + let key = key_of(&rule); + if let Some(e) = self.active.get_mut(&key) { + // Already active: upgrade to Manual + cancel any pending clear. + e.origin = BlackholeOrigin::Manual; + e.clear_requested_at = None; + return Vec::new(); + } + let target = rule.dst.addr(); + self.insert_rule(target, rule, now, BlackholeOrigin::Manual) + } + + /// Manually withdraw a rule (bypasses hold-down — an operator action is deliberate). + /// + /// # Arguments + /// + /// * `rule` - Identifies the rule to remove by its `(dst, protocol, dst_port)`. + /// + /// # Returns + /// + /// A `Withdraw` action if the rule was active, empty vector otherwise. + pub fn manual_remove(&mut self, rule: FlowSpecRule) -> Vec { + let key = key_of(&rule); + if let Some(e) = self.active.remove(&key) { + vec![FlowSpecAction::Withdraw(e.rule)] + } else { + Vec::new() + } + } + + /// Process time-driven withdrawals: deferred clears whose hold-down has now + /// elapsed, and auto rules past their TTL. Call periodically. + pub fn tick(&mut self, now: u64) -> Vec { + let hold_ms = u64::try_from(self.config.hold_down.as_millis()).unwrap_or(u64::MAX); + let ttl_ms = self + .config + .max_ttl + .map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)); + let mut expired: Vec = Vec::new(); + for (key, e) in &self.active { + let cleared_due = e + .clear_requested_at + .is_some_and(|_| now.saturating_sub(e.announced_at) >= hold_ms); + let ttl_due = matches!(e.origin, BlackholeOrigin::Auto) + && ttl_ms.is_some_and(|ttl| now.saturating_sub(e.last_activity) >= ttl); + if cleared_due || ttl_due { + expired.push(*key); + } + } + expired + .into_iter() + .filter_map(|k| { + self.active + .remove(&k) + .map(|e| FlowSpecAction::Withdraw(e.rule)) + }) + .collect() + } + + /// Re-install a persisted FlowSpec rule on a fresh session (rehydration). + pub fn resume( + &mut self, + rule: FlowSpecRule, + announced_at: u64, + origin: BlackholeOrigin, + ) -> Vec { + let target = rule.dst.addr(); + self.insert_rule(target, rule, announced_at, origin) + } + + /// Snapshot the active set (for reconcile mirroring, `list`, and tests). + #[must_use] + pub fn active_rules(&self) -> Vec<(FlowKey, u64, BlackholeOrigin)> { + self.active + .iter() + .map(|(k, e)| (*k, e.announced_at, e.origin)) + .collect() + } + + /// Re-assert every active rule for `target` without re-announcing. + /// + /// Used for [`blackwall_flow::FlowMitigationEvent::Update`], which reports that + /// an attack is still ongoing but (unlike `Open`) carries no rule list to + /// re-install. Cancels any pending deferred clear and refreshes the TTL + /// anchor on every active rule of `target`'s host route, mirroring the + /// re-assertion branch of `Self::insert_rule` — but for rules already known + /// to the controller, so no BGP re-announce or journal write is needed. + pub fn refresh_target(&mut self, target: IpAddr, now: u64) { + let host = host_prefix(target); + for (key, entry) in &mut self.active { + if key.0 == host { + entry.clear_requested_at = None; + entry.last_activity = now; + } + } + } + + /// Whether `target`'s host route falls inside a configured eligible prefix. + /// + /// Pure accessor over [`FlowSpecConfig::eligible_prefixes`]; lets a caller + /// (e.g. the manager) classify a rejected `manual_add`/`resume` without + /// duplicating the controller's eligibility logic. + #[must_use] + pub fn is_eligible(&self, target: IpAddr) -> bool { + self.config + .eligible_prefixes + .iter() + .any(|p| p.contains(&target)) + } + + fn insert_rule( + &mut self, + target: IpAddr, + rule: FlowSpecRule, + announced_at: u64, + origin: BlackholeOrigin, + ) -> Vec { + if !self.is_eligible(target) { + tracing::warn!(%target, "FlowSpec: target outside eligible prefixes; ignoring"); + return Vec::new(); + } + let key = key_of(&rule); + if let Some(e) = self.active.get_mut(&key) { + // Re-assertion of an already-active rule (e.g. a re-attack during a + // deferred-clear window): cancel any pending clear and refresh the + // TTL anchor so `tick` does not withdraw a flow under attack again. + e.clear_requested_at = None; + e.last_activity = announced_at; + return Vec::new(); + } + if self.active.len() >= self.config.max_rules { + tracing::warn!(%target, cap = self.config.max_rules, "FlowSpec: at cap; ignoring"); + return Vec::new(); + } + self.active.insert( + key, + ActiveEntry { + rule: rule.clone(), + announced_at, + last_activity: announced_at, + origin, + clear_requested_at: None, + }, + ); + vec![FlowSpecAction::Announce(rule)] + } +} + +/// Derive the `FlowKey` a rule is stored/looked-up under. +pub(crate) fn key_of(rule: &FlowSpecRule) -> FlowKey { + ( + rule.dst, + rule.protocol.unwrap_or(0), + rule.dst_port.unwrap_or(0), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::IpAddr; + use std::time::Duration; + + fn cfg() -> FlowSpecConfig { + FlowSpecConfig { + eligible_prefixes: vec!["203.0.113.0/24".parse().unwrap()], + max_rules: 3, + hold_down: Duration::from_secs(10), + max_ttl: None, + } + } + + fn cfg_ttl(ms: u64) -> FlowSpecConfig { + FlowSpecConfig { + max_ttl: Some(Duration::from_millis(ms)), + ..cfg() + } + } + + fn net(s: &str) -> IpNet { + s.parse().unwrap() + } + + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + + fn rule(dst: &str, protocol: u8, dst_port: u16, rate: f32) -> FlowSpecRule { + FlowSpecRule { + dst: net(dst), + protocol: Some(protocol), + dst_port: Some(dst_port), + action: FlowAction::TrafficRate(rate), + } + } + + #[test] + fn install_eligible_announces_rule_per_flow() { + let mut c = FlowSpecController::new(cfg()); + let actions = c.install(ip("203.0.113.7"), &[(17, 53, 1000.0), (6, 80, 500.0)], 1000); + assert_eq!(actions.len(), 2); + for a in &actions { + let FlowSpecAction::Announce(r) = a else { + panic!("expected Announce") + }; + assert_eq!(r.dst, net("203.0.113.7/32")); + } + } + + #[test] + fn install_ineligible_is_ignored() { + let mut c = FlowSpecController::new(cfg()); + assert!(c + .install(ip("198.51.100.7"), &[(17, 53, 1000.0)], 0) + .is_empty()); + } + + #[test] + fn clear_target_withdraws_auto_flow_but_keeps_manual_flow() { + // A target with BOTH an Auto and a Manual flow: clear_target (past + // hold-down) withdraws the Auto flow but never the operator's Manual one — + // each of a target's flows is cleared per its own origin. + let mut c = FlowSpecController::new(cfg()); + c.install(ip("203.0.113.7"), &[(17, 53, 0.0)], 0); // Auto + c.manual_add(rule("203.0.113.7/32", 6, 80, 0.0), 0); // Manual, same target + let actions = c.clear_target(ip("203.0.113.7"), 20_000); // past 10s hold-down + assert_eq!(actions.len(), 1, "only the Auto flow is withdrawn"); + let FlowSpecAction::Withdraw(r) = &actions[0] else { + panic!("expected Withdraw") + }; + assert_eq!((r.protocol, r.dst_port), (Some(17), Some(53))); + // the Manual flow survives. + let remaining = c.active_rules(); + assert_eq!(remaining.len(), 1); + assert_eq!((remaining[0].0 .1, remaining[0].0 .2), (6, 80)); + } + + #[test] + fn at_cap_is_ignored_across_total_rules() { + let mut c = FlowSpecController::new(cfg()); // max_rules 3 + let a = c.install( + ip("203.0.113.1"), + &[(17, 53, 1.0), (6, 80, 1.0), (6, 443, 1.0)], + 0, + ); + assert_eq!(a.len(), 3); + let b = c.install(ip("203.0.113.2"), &[(17, 53, 1.0)], 0); + assert!(b.is_empty(), "at cap across all targets' rules"); + } + + #[test] + fn clear_target_defers_then_tick_withdraws_all_flows() { + let mut c = FlowSpecController::new(cfg()); // 10s hold-down + c.install(ip("203.0.113.7"), &[(17, 53, 1.0), (6, 80, 1.0)], 0); + assert!(c.clear_target(ip("203.0.113.7"), 5_000).is_empty()); + assert!(c.tick(9_000).is_empty()); + let actions = c.tick(10_000); + assert_eq!(actions.len(), 2, "tick withdraws all of the target's flows"); + assert!(c.active_rules().is_empty()); + } + + #[test] + fn clear_target_at_or_after_hold_down_withdraws_immediately() { + let mut c = FlowSpecController::new(cfg()); + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + let actions = c.clear_target(ip("203.0.113.7"), 10_000); + assert_eq!(actions.len(), 1); + assert!(c.active_rules().is_empty()); + } + + #[test] + fn clear_target_does_not_affect_other_targets() { + let mut c = FlowSpecController::new(cfg()); + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + c.install(ip("203.0.113.8"), &[(17, 53, 1.0)], 0); + let actions = c.clear_target(ip("203.0.113.7"), 10_000); + assert_eq!(actions.len(), 1); + assert_eq!(c.active_rules().len(), 1); + } + + #[test] + fn clear_target_with_no_active_flows_is_empty() { + let mut c = FlowSpecController::new(cfg()); + assert!(c.clear_target(ip("203.0.113.7"), 0).is_empty()); + } + + #[test] + fn ttl_withdraws_a_never_cleared_auto_rule() { + let mut c = FlowSpecController::new(cfg_ttl(30_000)); // 30s TTL + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + assert!(c.tick(29_000).is_empty()); + assert_eq!(c.tick(30_000).len(), 1); + } + + #[test] + fn manual_survives_clear_target_and_ttl() { + let mut c = FlowSpecController::new(cfg_ttl(30_000)); + c.manual_add(rule("203.0.113.7/32", 17, 53, 1.0), 0); + assert!( + c.clear_target(ip("203.0.113.7"), 100_000).is_empty(), + "auto-clear must not withdraw a manual rule" + ); + assert!(c.tick(200_000).is_empty(), "and tick must not either"); + assert_eq!(c.active_rules().len(), 1); + } + + #[test] + fn reinstall_during_deferred_clear_cancels_withdraw() { + let mut c = FlowSpecController::new(cfg()); // 10s hold-down + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + assert!(c.clear_target(ip("203.0.113.7"), 5_000).is_empty()); + // Re-assertion (idempotent — no new announce) must re-arm the entry. + assert!(c + .install(ip("203.0.113.7"), &[(17, 53, 1.0)], 6_000) + .is_empty()); + assert!( + c.tick(10_000).is_empty(), + "re-assertion cancelled the deferred withdraw" + ); + assert_eq!(c.active_rules().len(), 1); + } + + #[test] + fn active_rules_snapshot() { + let mut c = FlowSpecController::new(cfg()); + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 1234); + let snap = c.active_rules(); + assert_eq!(snap.len(), 1); + assert_eq!(snap[0].0, (net("203.0.113.7/32"), 17, 53)); + assert_eq!(snap[0].1, 1234); + assert_eq!(snap[0].2, BlackholeOrigin::Auto); + } + + #[test] + fn resume_reannounces_and_counts_against_cap() { + let mut c = FlowSpecController::new(cfg()); // max_rules 3 + let r = rule("203.0.113.5/32", 17, 53, 1.0); + let actions = c.resume(r.clone(), 12_345, BlackholeOrigin::Manual); + assert_eq!(actions, vec![FlowSpecAction::Announce(r)]); + let snap = c.active_rules(); + assert_eq!(snap[0].1, 12_345); + assert_eq!(snap[0].2, BlackholeOrigin::Manual); + // Counts against the cap: 1 (resumed) + 2 more = 3 (cap); a 4th is rejected. + assert_eq!( + c.install(ip("203.0.113.6"), &[(6, 80, 1.0), (6, 443, 1.0)], 0) + .len(), + 2 + ); + assert!( + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0).is_empty(), + "at cap" + ); + } + + #[test] + fn manual_add_upgrades_auto_to_manual() { + let mut c = FlowSpecController::new(cfg()); + assert_eq!(c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0).len(), 1); // Auto + assert!( + c.manual_add(rule("203.0.113.7/32", 17, 53, 1.0), 1_000) + .is_empty(), + "already active -> no new announce" + ); + // Now an auto-clear can't remove it. + assert!(c.clear_target(ip("203.0.113.7"), 100_000).is_empty()); + assert!(c.tick(200_000).is_empty()); + let snap = c.active_rules(); + assert_eq!(snap.len(), 1); + assert_eq!(snap[0].2, BlackholeOrigin::Manual); + } + + #[test] + fn manual_remove_bypasses_hold_down() { + let mut c = FlowSpecController::new(cfg()); + let r = rule("203.0.113.7/32", 17, 53, 1.0); + c.manual_add(r.clone(), 0); + assert_eq!( + c.manual_remove(r.clone()), + vec![FlowSpecAction::Withdraw(r)] + ); + assert!(c.active_rules().is_empty()); + } + + #[test] + fn manual_remove_of_absent_rule_is_empty() { + let mut c = FlowSpecController::new(cfg()); + assert!(c + .manual_remove(rule("203.0.113.7/32", 17, 53, 1.0)) + .is_empty()); + } + + #[test] + fn ipv6_target_uses_128() { + let mut cfg6 = cfg(); + cfg6.eligible_prefixes = vec![net("2001:db8::/32")]; + let mut c = FlowSpecController::new(cfg6); + let actions = c.install(ip("2001:db8::7"), &[(17, 53, 1.0)], 0); + let FlowSpecAction::Announce(r) = &actions[0] else { + panic!("expected Announce") + }; + assert_eq!(r.dst, net("2001:db8::7/128")); + } + + #[test] + fn refresh_target_cancels_pending_clear_and_survives_tick() { + let mut c = FlowSpecController::new(cfg()); // 10s hold-down + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + assert!(c.clear_target(ip("203.0.113.7"), 5_000).is_empty()); + c.refresh_target(ip("203.0.113.7"), 6_000); + assert!( + c.tick(10_000).is_empty(), + "refresh_target cancelled the deferred withdraw" + ); + assert_eq!(c.active_rules().len(), 1); + } + + #[test] + fn refresh_target_does_not_affect_other_targets() { + let mut c = FlowSpecController::new(cfg()); + c.install(ip("203.0.113.7"), &[(17, 53, 1.0)], 0); + c.install(ip("203.0.113.8"), &[(17, 53, 1.0)], 0); + assert!(c.clear_target(ip("203.0.113.7"), 5_000).is_empty()); + assert!(c.clear_target(ip("203.0.113.8"), 5_000).is_empty()); + c.refresh_target(ip("203.0.113.7"), 6_000); + // only .7 was refreshed; .8's deferred clear still fires at hold-down. + let actions = c.tick(10_000); + assert_eq!(actions.len(), 1); + assert_eq!(c.active_rules().len(), 1); + } + + #[test] + fn is_eligible_checks_configured_prefixes() { + let c = FlowSpecController::new(cfg()); + assert!(c.is_eligible(ip("203.0.113.7"))); + assert!(!c.is_eligible(ip("198.51.100.7"))); + } +} diff --git a/crates/blackwall-rtbh/src/flowspec_manager.rs b/crates/blackwall-rtbh/src/flowspec_manager.rs new file mode 100644 index 0000000..7494569 --- /dev/null +++ b/crates/blackwall-rtbh/src/flowspec_manager.rs @@ -0,0 +1,835 @@ +//! Single-owner FlowSpec manager: executes controller decisions on BGP and +//! mirrors auto/manual rule state into a persistence journal. +//! +//! The FlowSpec analogue of [`crate::manager::RtbhManager`]: the pure +//! [`FlowSpecController`] decides; this module owns the I/O boundary (BGP +//! session + journal), reusing [`crate::manager::BgpExecutor`] (extended with +//! the FlowSpec announce/withdraw methods) and a dedicated [`FlowSpecJournal`] +//! seam so `blackwall-rtbh` stays free of any DB dependency. Every invariant +//! of `RtbhManager` carries over, adapted from a single blackholed IP to a +//! target's multiple `(protocol, port)` rules keyed by [`FlowKey`]: the +//! `pending_mirror` self-heal is coalesced by `FlowKey` instead of by target +//! IP, a fallible journal write keeps the rule active on failure, and a BGP +//! failure is never journaled (no phantom rule). + +use crate::controller::BlackholeOrigin; +use crate::flowspec_controller::{key_of, FlowKey, FlowSpecAction, FlowSpecController}; +use crate::manager::{ApplyOutcome, BgpExecutor, JournalError}; +use async_trait::async_trait; +use blackwall_bgp::FlowSpecRule; +use blackwall_flow::FlowRule; +use std::net::IpAddr; + +/// Mirrors FlowSpec rule state into persistent storage. +/// +/// This is the sole seam through which the FlowSpec side of `blackwall-rtbh` +/// would touch a database — the crate itself never depends on one. +/// Implemented elsewhere (e.g. the control-plane crate that owns the DB) and +/// injected here. See [`crate::manager::BlackholeJournal`] for the RTBH +/// analogue. +#[async_trait] +pub trait FlowSpecJournal: Send + Sync { + /// Record that `rule` is now announced, with the given origin. + async fn record_announce( + &self, + rule: FlowSpecRule, + origin: BlackholeOrigin, + at_ms: u64, + ) -> Result<(), JournalError>; + /// Record that `rule` is no longer announced. + async fn record_withdraw(&self, rule: FlowSpecRule, at_ms: u64) -> Result<(), JournalError>; +} + +/// A journal mirror write that failed and is queued for a self-heal retry. +/// +/// The BGP side of the operation already succeeded when this is queued, so +/// retrying only ever re-attempts the journal write — never BGP. Mirrors +/// [`crate::manager`]'s private `MirrorOp`, keyed by [`FlowKey`] instead of +/// target IP (a target may have several concurrently-queued rules). +#[derive(Debug, Clone, PartialEq)] +enum MirrorOp { + /// Re-attempt `record_announce` for `rule`. + Announce { + rule: FlowSpecRule, + origin: BlackholeOrigin, + at_ms: u64, + }, + /// Re-attempt `record_withdraw` for `rule`. + Withdraw { rule: FlowSpecRule, at_ms: u64 }, +} + +impl MirrorOp { + /// The `FlowKey` this mirror op concerns. + fn key(&self) -> FlowKey { + match self { + MirrorOp::Announce { rule, .. } | MirrorOp::Withdraw { rule, .. } => key_of(rule), + } + } +} + +/// Single-owner FlowSpec manager. +/// +/// Owns the pure [`FlowSpecController`] plus the I/O boundary: it executes the +/// controller's decisions on a [`BgpExecutor`] and mirrors auto/manual rule +/// state via a [`FlowSpecJournal`]. A BGP failure is logged and the action is +/// not journaled — but note this is a known limitation, not a retry +/// mechanism: on a failed first announce the controller entry is kept in +/// memory while the rule itself is never re-announced automatically. A +/// journal failure after a successful BGP operation is logged, never causes a +/// live rule to be withdrawn, and is queued as a `MirrorOp` for a bounded +/// self-heal retry on the next [`FlowSpecManager::tick`] — the BGP outcome is +/// never re-issued, only the mirror write. +pub struct FlowSpecManager { + controller: FlowSpecController, + bgp: B, + journal: J, + /// Journal writes that failed after their BGP operation already + /// succeeded; retried (never re-issued to BGP) by + /// `FlowSpecManager::retry_pending_mirror` on the next tick. + pending_mirror: Vec, +} + +impl FlowSpecManager { + /// Wrap a controller with a BGP executor and a journal. + pub fn new(controller: FlowSpecController, bgp: B, journal: J) -> Self { + Self { + controller, + bgp, + journal, + pending_mirror: Vec::new(), + } + } + + /// Install the flow-scoped rules selected for `target` (from a + /// [`blackwall_flow::FlowMitigationEvent::Open`]) and execute + journal + /// the resulting announces. + /// + /// Announces are journaled as [`BlackholeOrigin::Auto`] (the only origin + /// this path can produce). A BGP error is logged and the action is not + /// journaled. A journal error after a successful BGP operation is logged + /// and queued for a self-heal retry on the next tick (the controller + /// entry is kept — never withdraw a live rule because the DB write + /// failed). + pub async fn apply_open( + &mut self, + target: IpAddr, + rules: &[FlowRule], + mono_now: u64, + wall_now: u64, + ) { + let tuples: Vec<(u8, u16, f32)> = rules + .iter() + .map(|r| (r.proto, r.dst_port, r.rate)) + .collect(); + let actions = self.controller.install(target, &tuples, mono_now); + for action in actions { + self.execute_and_journal(action, wall_now).await; + } + } + + /// Clear every active rule for `target` (from a + /// [`blackwall_flow::FlowMitigationEvent::Clear`]) and execute + journal + /// the withdraws that clear immediately (hold-down permitting; the rest + /// are deferred to a later [`Self::tick`]). + pub async fn apply_clear(&mut self, target: IpAddr, mono_now: u64, wall_now: u64) { + let actions = self.controller.clear_target(target, mono_now); + for action in actions { + self.execute_and_journal(action, wall_now).await; + } + } + + /// Re-assert `target`'s active rules (from a + /// [`blackwall_flow::FlowMitigationEvent::Update`]), cancelling any + /// pending deferred clear and refreshing the TTL anchor. + /// + /// `Update` carries only the target IP (not the concrete rules), so unlike + /// `apply_open` this cannot re-run `install`; it instead calls + /// [`FlowSpecController::refresh_target`], a minimal target-refresh entry + /// point added for this purpose. No BGP call or journal write is needed — + /// the rule is already announced and already mirrored; only the + /// controller's in-memory bookkeeping changes. + pub fn apply_updated(&mut self, target: IpAddr, mono_now: u64) { + self.controller.refresh_target(target, mono_now); + } + + /// Process time-driven withdrawals (deferred clears, TTL expiry) and + /// execute + journal each one. + /// + /// Starts by retrying any journal mirror writes queued by a previous + /// tick's transient failure (see + /// `FlowSpecManager::retry_pending_mirror`), so a self-heal converges + /// within one tick interval of the DB recovering. + pub async fn tick(&mut self, mono_now: u64, wall_now: u64) { + self.retry_pending_mirror().await; + let actions = self.controller.tick(mono_now); + for action in actions { + self.execute_and_journal(action, wall_now).await; + } + } + + /// Manually install a FlowSpec rule. + /// + /// Returns [`ApplyOutcome::Applied`] if newly installed or upgraded from + /// `Auto` to `Manual` (re-journaled as `Manual` in the latter case), + /// [`ApplyOutcome::Deferred`] if the manager is at capacity, or + /// [`ApplyOutcome::Rejected`] if the target is ineligible. Unlike RTBH, + /// FlowSpec carries no next-hop, so there is no next-hop rejection case. + pub async fn apply_add( + &mut self, + rule: FlowSpecRule, + mono_now: u64, + wall_now: u64, + ) -> ApplyOutcome { + let key = key_of(&rule); + let target = rule.dst.addr(); + let actions = self.controller.manual_add(rule.clone(), mono_now); + if let Some(FlowSpecAction::Announce(r)) = actions.into_iter().next() { + self.execute_and_journal_announce(r, BlackholeOrigin::Manual, wall_now) + .await; + return ApplyOutcome::Applied; + } + // Empty result: either already active (upgrade), at cap, or rejected. + if self.is_active(key) { + // Upgrade: promote the mirror to Manual. + if let Err(e) = self + .journal + .record_announce(rule.clone(), BlackholeOrigin::Manual, wall_now) + .await + { + tracing::error!(%target, error = %e, "FlowSpec: journal write failed after manual upgrade; keeping active"); + self.queue_mirror(MirrorOp::Announce { + rule, + origin: BlackholeOrigin::Manual, + at_ms: wall_now, + }); + } + return ApplyOutcome::Applied; + } + if !self.controller.is_eligible(target) { + return ApplyOutcome::Rejected(format!("{target} is outside eligible prefixes")); + } + ApplyOutcome::Deferred + } + + /// Manually withdraw a rule (bypasses hold-down). + pub async fn apply_remove(&mut self, rule: FlowSpecRule, wall_now: u64) { + let actions = self.controller.manual_remove(rule); + for action in actions { + self.execute_and_journal(action, wall_now).await; + } + } + + /// Re-install persisted FlowSpec rules on a fresh session (rehydration). + /// + /// For each row, calls [`FlowSpecController::resume`] and re-announces on + /// BGP (without journaling — the row already exists in the journal). If + /// `resume` returns no action (over cap or ineligible), this logs a + /// warning naming the target; a row is never silently dropped. + pub async fn rehydrate( + &mut self, + rows: Vec<(FlowSpecRule, u64, BlackholeOrigin)>, + mono_now: u64, + ) { + for (rule, _persisted_at, origin) in rows { + let target = rule.dst.addr(); + let actions = self.controller.resume(rule.clone(), mono_now, origin); + if let Some(FlowSpecAction::Announce(r)) = actions.into_iter().next() { + if let Err(e) = self.bgp.announce_flowspec(r).await { + tracing::warn!(%target, error = %e, "FlowSpec: rehydrate re-announce failed"); + } + continue; + } + // resume() returned nothing: over cap or ineligible. A persisted + // row must never be silently dropped — always warn. + let reason = if self.controller.is_eligible(target) { + "at cap" + } else { + "ineligible" + }; + tracing::warn!(%target, reason, "FlowSpec: rehydrate dropped a persisted rule"); + } + } + + /// Snapshot the active set (for reconcile mirroring, `list`, and tests). + #[must_use] + pub fn active(&self) -> Vec<(FlowKey, u64, BlackholeOrigin)> { + self.controller.active_rules() + } + + fn is_active(&self, key: FlowKey) -> bool { + self.controller + .active_rules() + .iter() + .any(|(k, ..)| *k == key) + } + + /// Queue a failed mirror write for self-heal, coalescing by [`FlowKey`]. + /// + /// The mirror only needs to reflect the current active set, so keeping just + /// the latest op per key is both correct (journal ops converge to a final + /// state) and bounds the queue to one entry per rule — a rule that flaps + /// while the DB is down can never grow the queue without bound. + fn queue_mirror(&mut self, op: MirrorOp) { + let key = op.key(); + self.pending_mirror.retain(|o| o.key() != key); + self.pending_mirror.push(op); + } + + /// Execute one controller action on BGP and mirror it into the journal. + async fn execute_and_journal(&mut self, action: FlowSpecAction, wall_now: u64) { + match action { + FlowSpecAction::Announce(rule) => { + self.execute_and_journal_announce(rule, BlackholeOrigin::Auto, wall_now) + .await; + } + FlowSpecAction::Withdraw(rule) => { + let key = key_of(&rule); + if let Err(e) = self.bgp.withdraw_flowspec(rule.clone()).await { + tracing::warn!(?key, error = %e, "FlowSpec: BGP withdraw failed; not journaling"); + return; + } + if let Err(e) = self.journal.record_withdraw(rule.clone(), wall_now).await { + tracing::error!(?key, error = %e, "FlowSpec: journal withdraw-mirror failed; rule already withdrawn from BGP (mirror row will be stale)"); + self.queue_mirror(MirrorOp::Withdraw { + rule, + at_ms: wall_now, + }); + } + } + } + } + + async fn execute_and_journal_announce( + &mut self, + rule: FlowSpecRule, + origin: BlackholeOrigin, + wall_now: u64, + ) { + let key = key_of(&rule); + if let Err(e) = self.bgp.announce_flowspec(rule.clone()).await { + tracing::warn!(?key, error = %e, "FlowSpec: BGP announce failed; not journaling"); + return; + } + if let Err(e) = self + .journal + .record_announce(rule.clone(), origin, wall_now) + .await + { + tracing::error!(?key, error = %e, "FlowSpec: journal write failed after announce; keeping active"); + self.queue_mirror(MirrorOp::Announce { + rule, + origin, + at_ms: wall_now, + }); + } + } + + /// Drain-retry queued mirror writes left over from a transient journal + /// failure. + /// + /// The BGP side of each queued op already succeeded when it was queued, + /// so this only ever re-attempts the matching journal call — it never + /// re-announces or re-withdraws on BGP. Ops that still fail are kept + /// (retried again on the next call); ops that succeed are dropped. + /// Queued ops are retried in order, so an Announce followed by a later + /// Withdraw for the same key converge correctly. + async fn retry_pending_mirror(&mut self) { + if self.pending_mirror.is_empty() { + return; + } + let ops = std::mem::take(&mut self.pending_mirror); + for op in ops { + let result = match &op { + MirrorOp::Announce { + rule, + origin, + at_ms, + } => { + self.journal + .record_announce(rule.clone(), *origin, *at_ms) + .await + } + MirrorOp::Withdraw { rule, at_ms } => { + self.journal.record_withdraw(rule.clone(), *at_ms).await + } + }; + if let Err(e) = result { + tracing::warn!(op = ?op, error = %e, "FlowSpec: mirror self-heal retry failed; re-queuing"); + self.pending_mirror.push(op); + } + } + } + + #[cfg(test)] + pub(crate) fn bgp(&self) -> &B { + &self.bgp + } + + /// Number of journal mirror writes currently queued for self-heal retry. + #[cfg(test)] + pub(crate) fn pending_mirror_len(&self) -> usize { + self.pending_mirror.len() + } + + #[cfg(test)] + pub(crate) fn journal(&self) -> &J { + &self.journal + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::flowspec_controller::FlowSpecConfig; + use blackwall_bgp::FlowAction; + use std::net::IpAddr; + use std::sync::Mutex; + use std::time::Duration; + + #[derive(Default)] + struct FakeBgp { + announced: Mutex>, + withdrawn: Mutex>, + fail: bool, + } + #[async_trait] + impl BgpExecutor for FakeBgp { + async fn announce( + &self, + _route: blackwall_bgp::Route, + ) -> Result<(), crate::manager::BgpError> { + unreachable!("FlowSpecManager never calls the RTBH side of BgpExecutor") + } + async fn withdraw(&self, _prefix: ipnet::IpNet) -> Result<(), crate::manager::BgpError> { + unreachable!("FlowSpecManager never calls the RTBH side of BgpExecutor") + } + async fn announce_flowspec( + &self, + rule: FlowSpecRule, + ) -> Result<(), crate::manager::BgpError> { + if self.fail { + return Err(crate::manager::BgpError); + } + self.announced.lock().unwrap().push(rule); + Ok(()) + } + async fn withdraw_flowspec( + &self, + rule: FlowSpecRule, + ) -> Result<(), crate::manager::BgpError> { + if self.fail { + return Err(crate::manager::BgpError); + } + self.withdrawn.lock().unwrap().push(rule); + Ok(()) + } + } + + #[derive(Default)] + struct FakeJournal { + announced: Mutex>, + withdrawn: Mutex>, + fail: bool, + /// Number of upcoming calls (announce or withdraw, whichever comes + /// first) that should fail before the journal starts succeeding — + /// simulates a transient DB blip that self-heals. + fail_calls_remaining: Mutex, + } + #[async_trait] + impl FlowSpecJournal for FakeJournal { + async fn record_announce( + &self, + rule: FlowSpecRule, + origin: BlackholeOrigin, + _at: u64, + ) -> Result<(), JournalError> { + if self.fail || self.take_transient_failure() { + return Err(JournalError("boom".into())); + } + self.announced.lock().unwrap().push((rule, origin)); + Ok(()) + } + async fn record_withdraw(&self, rule: FlowSpecRule, _at: u64) -> Result<(), JournalError> { + if self.fail || self.take_transient_failure() { + return Err(JournalError("boom".into())); + } + self.withdrawn.lock().unwrap().push(rule); + Ok(()) + } + } + impl FakeJournal { + /// Consume one remaining scheduled transient failure, if any. + fn take_transient_failure(&self) -> bool { + let mut remaining = self.fail_calls_remaining.lock().unwrap(); + if *remaining > 0 { + *remaining -= 1; + true + } else { + false + } + } + } + + fn cfg() -> FlowSpecConfig { + FlowSpecConfig { + eligible_prefixes: vec!["203.0.113.0/24".parse().unwrap()], + max_rules: 2, + hold_down: Duration::from_secs(10), + max_ttl: None, + } + } + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + fn flow_rule(dst: &str, proto: u8, dst_port: u16, rate: f32) -> FlowRule { + FlowRule { + dst: ip(dst), + proto, + dst_port, + rate, + } + } + fn rule(dst: &str, protocol: u8, dst_port: u16, rate: f32) -> FlowSpecRule { + FlowSpecRule { + dst: dst.parse().unwrap(), + protocol: Some(protocol), + dst_port: Some(dst_port), + action: FlowAction::TrafficRate(rate), + } + } + fn mgr(fail_bgp: bool, fail_j: bool) -> FlowSpecManager { + FlowSpecManager::new( + FlowSpecController::new(cfg()), + FakeBgp { + fail: fail_bgp, + ..Default::default() + }, + FakeJournal { + fail: fail_j, + ..Default::default() + }, + ) + } + + /// A manager whose journal fails its first `n` calls (BGP transient + /// blip), then succeeds — used to exercise the mirror self-heal retry. + fn mgr_transient_journal_failures(n: usize) -> FlowSpecManager { + FlowSpecManager::new( + FlowSpecController::new(cfg()), + FakeBgp::default(), + FakeJournal { + fail_calls_remaining: Mutex::new(n), + ..Default::default() + }, + ) + } + + #[tokio::test] + async fn apply_open_announces_and_journals_each_rule() { + let mut m = mgr(false, false); + m.apply_open( + ip("203.0.113.7"), + &[ + flow_rule("203.0.113.7", 17, 53, 0.0), + flow_rule("203.0.113.7", 6, 80, 500.0), + ], + 0, + 5000, + ) + .await; + assert_eq!(m.active().len(), 2); + assert_eq!(m.bgp().announced.lock().unwrap().len(), 2); + assert_eq!(m.journal().announced.lock().unwrap().len(), 2); + for (_, origin) in m.journal().announced.lock().unwrap().iter() { + assert_eq!(*origin, BlackholeOrigin::Auto); + } + } + + #[tokio::test] + async fn apply_clear_withdraws_all_target_flows() { + let mut m = mgr(false, false); + m.apply_open( + ip("203.0.113.7"), + &[ + flow_rule("203.0.113.7", 17, 53, 0.0), + flow_rule("203.0.113.7", 6, 80, 0.0), + ], + 0, + 0, + ) + .await; + // past the 10s hold-down: withdraws immediately. + m.apply_clear(ip("203.0.113.7"), 10_000, 10_000).await; + assert!(m.active().is_empty()); + assert_eq!(m.bgp().withdrawn.lock().unwrap().len(), 2); + assert_eq!(m.journal().withdrawn.lock().unwrap().len(), 2); + } + + #[tokio::test] + async fn apply_clear_defers_within_hold_down_then_tick_withdraws() { + let mut m = mgr(false, false); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + m.apply_clear(ip("203.0.113.7"), 5000, 0).await; + assert_eq!(m.active().len(), 1, "deferred, not yet withdrawn"); + m.tick(10_000, 0).await; + assert!(m.active().is_empty(), "tick withdraws after hold-down"); + } + + #[tokio::test] + async fn apply_updated_cancels_pending_clear() { + let mut m = mgr(false, false); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + m.apply_clear(ip("203.0.113.7"), 5000, 0).await; + assert_eq!(m.active().len(), 1, "deferred, not yet withdrawn"); + m.apply_updated(ip("203.0.113.7"), 6000); + m.tick(10_000, 0).await; + assert_eq!( + m.active().len(), + 1, + "Update cancelled the pending clear before the deferred hold-down elapsed" + ); + } + + #[tokio::test] + async fn journal_failure_keeps_active_and_queues_pending_mirror() { + let mut m = mgr(false, true); // journal fails + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + assert_eq!( + m.active().len(), + 1, + "a journal error must not drop a live FlowSpec rule" + ); + assert!( + m.journal().announced.lock().unwrap().is_empty(), + "the failed announce must not have been recorded" + ); + assert_eq!( + m.pending_mirror_len(), + 1, + "the failed mirror write must be queued for self-heal" + ); + } + + #[tokio::test] + async fn bgp_failure_does_not_journal() { + let mut m = mgr(true, false); // BGP fails + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + assert!( + m.journal().announced.lock().unwrap().is_empty(), + "a BGP failure must not be journaled (no phantom rule)" + ); + assert!( + m.bgp().announced.lock().unwrap().is_empty(), + "and of course BGP itself recorded nothing" + ); + } + + #[tokio::test] + async fn tick_drains_pending_mirror_once_journal_recovers() { + let mut m = mgr_transient_journal_failures(1); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 1234, + ) + .await; + assert_eq!(m.pending_mirror_len(), 1); + assert!(m.journal().announced.lock().unwrap().is_empty()); + + m.tick(1000, 5000).await; + + assert_eq!(m.pending_mirror_len(), 0); + assert_eq!(m.journal().announced.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn retry_pending_mirror_requeues_on_repeat_failure() { + let mut m = mgr(false, true); // journal always fails + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 1234, + ) + .await; + assert_eq!(m.pending_mirror_len(), 1); + + m.tick(1000, 5000).await; + + assert_eq!( + m.pending_mirror_len(), + 1, + "a still-failing journal must keep the op queued, not drop it" + ); + assert!(m.journal().announced.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn apply_add_then_remove() { + let mut m = mgr(false, false); + let r = rule("203.0.113.7/32", 17, 53, 0.0); + assert_eq!(m.apply_add(r.clone(), 0, 0).await, ApplyOutcome::Applied); + assert_eq!(m.active().len(), 1); + assert_eq!( + m.journal().announced.lock().unwrap()[0].1, + BlackholeOrigin::Manual + ); + m.apply_remove(r, 1000).await; + assert!(m.active().is_empty()); + assert_eq!(m.journal().withdrawn.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn apply_add_upgrade_rejournals_as_manual() { + let mut m = mgr(false, false); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + let r = rule("203.0.113.7/32", 17, 53, 0.0); + assert_eq!(m.apply_add(r, 1000, 2000).await, ApplyOutcome::Applied); + let recorded = m.journal().announced.lock().unwrap(); + assert_eq!(recorded.len(), 2); + assert_eq!(recorded[0].1, BlackholeOrigin::Auto); + assert_eq!(recorded[1].1, BlackholeOrigin::Manual); + } + + #[tokio::test] + async fn apply_add_rejects_ineligible_and_defers_at_cap() { + let mut m = mgr(false, false); // cap = 2 + assert!(matches!( + m.apply_add(rule("198.51.100.9/32", 17, 53, 0.0), 0, 0) + .await, + ApplyOutcome::Rejected(_) + )); + assert_eq!( + m.apply_add(rule("203.0.113.1/32", 17, 53, 0.0), 0, 0).await, + ApplyOutcome::Applied + ); + assert_eq!( + m.apply_add(rule("203.0.113.1/32", 6, 80, 0.0), 0, 0).await, + ApplyOutcome::Applied + ); // cap=2 + assert_eq!( + m.apply_add(rule("203.0.113.1/32", 6, 443, 0.0), 0, 0).await, + ApplyOutcome::Deferred + ); + } + + #[tokio::test] + async fn rehydrate_reannounces() { + let mut m = mgr(false, false); + m.rehydrate( + vec![( + rule("203.0.113.5/32", 17, 53, 0.0), + 111, + BlackholeOrigin::Manual, + )], + 9000, + ) + .await; + assert_eq!(m.active().len(), 1); + assert_eq!(m.bgp().announced.lock().unwrap().len(), 1); + // rehydrate never journals — the row already exists. + assert!(m.journal().announced.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn rehydrate_warns_and_does_not_drop_silently_when_ineligible() { + let mut m = mgr(false, false); + m.rehydrate( + vec![( + rule("198.51.100.9/32", 17, 53, 0.0), + 111, + BlackholeOrigin::Manual, + )], + 9000, + ) + .await; + assert!(m.active().is_empty()); + } + + #[tokio::test] + async fn queue_mirror_coalesces_repeated_failures_for_one_key() { + // A single rule flapping while the journal is down must never grow + // the queue past one entry for that key. + let mut m = mgr(false, true); // BGP ok, journal always fails + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 1000, + ) + .await; + m.apply_remove(rule("203.0.113.7/32", 17, 53, 0.0), 2000) + .await; + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 3000, + 3000, + ) + .await; + assert_eq!( + m.pending_mirror_len(), + 1, + "repeated failures for one key coalesce to a single queued op" + ); + } + + #[tokio::test] + async fn queued_announce_then_withdraw_for_same_key_coalesces_to_withdraw() { + let mut m = mgr_transient_journal_failures(2); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 1000, + ) + .await; + assert_eq!(m.pending_mirror_len(), 1); + + m.apply_remove(rule("203.0.113.7/32", 17, 53, 0.0), 2000) + .await; + assert_eq!( + m.pending_mirror_len(), + 1, + "the withdraw coalesces with the queued announce for the same key" + ); + assert!(m.active().is_empty(), "BGP withdraw must still take effect"); + + m.tick(3000, 4000).await; + + assert_eq!(m.pending_mirror_len(), 0); + assert!(m.journal().announced.lock().unwrap().is_empty()); + assert_eq!(m.journal().withdrawn.lock().unwrap().len(), 1); + } +} diff --git a/crates/blackwall-rtbh/src/lib.rs b/crates/blackwall-rtbh/src/lib.rs index 6fffe29..3068bcb 100644 --- a/crates/blackwall-rtbh/src/lib.rs +++ b/crates/blackwall-rtbh/src/lib.rs @@ -4,9 +4,13 @@ //! that executes its decisions. pub mod controller; +pub mod flowspec_controller; +pub mod flowspec_manager; pub mod manager; pub use controller::{BlackholeOrigin, RtbhAction, RtbhConfig, RtbhController}; +pub use flowspec_controller::{FlowKey, FlowSpecAction, FlowSpecConfig, FlowSpecController}; +pub use flowspec_manager::{FlowSpecJournal, FlowSpecManager}; pub use manager::{ ApplyOutcome, BgpError, BgpExecutor, BlackholeJournal, JournalError, RtbhManager, }; @@ -24,4 +28,20 @@ impl manager::BgpExecutor for blackwall_bgp::BgpHandle { .await .map_err(Into::into) } + async fn announce_flowspec( + &self, + rule: blackwall_bgp::FlowSpecRule, + ) -> Result<(), manager::BgpError> { + blackwall_bgp::BgpHandle::announce_flowspec(self, rule) + .await + .map_err(Into::into) + } + async fn withdraw_flowspec( + &self, + rule: blackwall_bgp::FlowSpecRule, + ) -> Result<(), manager::BgpError> { + blackwall_bgp::BgpHandle::withdraw_flowspec(self, rule) + .await + .map_err(Into::into) + } } diff --git a/crates/blackwall-rtbh/src/manager.rs b/crates/blackwall-rtbh/src/manager.rs index df47502..7264126 100644 --- a/crates/blackwall-rtbh/src/manager.rs +++ b/crates/blackwall-rtbh/src/manager.rs @@ -22,6 +22,10 @@ pub trait BgpExecutor: Send + Sync { async fn announce(&self, route: Route) -> Result<(), BgpError>; /// Withdraw a previously-announced blackhole prefix. async fn withdraw(&self, prefix: IpNet) -> Result<(), BgpError>; + /// Announce a FlowSpec traffic-filter rule. + async fn announce_flowspec(&self, rule: blackwall_bgp::FlowSpecRule) -> Result<(), BgpError>; + /// Withdraw a previously-announced FlowSpec rule. + async fn withdraw_flowspec(&self, rule: blackwall_bgp::FlowSpecRule) -> Result<(), BgpError>; } /// Mirrors blackhole state into persistent storage. @@ -78,7 +82,7 @@ pub enum ApplyOutcome { /// on a failed first announce the controller entry is kept in memory while /// the route itself is never re-announced automatically. A journal failure /// after a successful BGP operation is logged, never causes a live -/// blackhole to be withdrawn, and is queued as a [`MirrorOp`] for a bounded +/// blackhole to be withdrawn, and is queued as a `MirrorOp` for a bounded /// self-heal retry on the next [`RtbhManager::tick`] — the BGP outcome is /// never re-issued, only the mirror write. pub struct RtbhManager { @@ -87,7 +91,7 @@ pub struct RtbhManager { journal: J, /// Journal writes that failed after their BGP operation already /// succeeded; retried (never re-issued to BGP) by - /// [`RtbhManager::retry_pending_mirror`] on the next tick. + /// `RtbhManager::retry_pending_mirror` on the next tick. pending_mirror: Vec, } @@ -147,7 +151,7 @@ impl RtbhManager { /// execute + journal each one. /// /// Starts by retrying any journal mirror writes queued by a previous - /// tick's transient failure (see [`RtbhManager::retry_pending_mirror`]), + /// tick's transient failure (see `RtbhManager::retry_pending_mirror`), /// so a self-heal converges within one tick interval of the DB /// recovering. pub async fn tick(&mut self, mono_now: u64, wall_now: u64) { @@ -401,6 +405,28 @@ mod tests { self.withdrawn.lock().unwrap().push(prefix); Ok(()) } + // RtbhManager never calls the FlowSpec side of BgpExecutor; these two + // arms exist only so this fake still implements the (now-shared) + // trait. FlowSpecManager's own tests exercise a dedicated fake that + // records these calls. + async fn announce_flowspec( + &self, + _rule: blackwall_bgp::FlowSpecRule, + ) -> Result<(), BgpError> { + if self.fail { + return Err(BgpError); + } + Ok(()) + } + async fn withdraw_flowspec( + &self, + _rule: blackwall_bgp::FlowSpecRule, + ) -> Result<(), BgpError> { + if self.fail { + return Err(BgpError); + } + Ok(()) + } } #[derive(Default)] struct FakeJournal { diff --git a/crates/blackwall-rtbh/tests/flowspec_auto_interop.rs b/crates/blackwall-rtbh/tests/flowspec_auto_interop.rs new file mode 100644 index 0000000..d9c49e6 --- /dev/null +++ b/crates/blackwall-rtbh/tests/flowspec_auto_interop.rs @@ -0,0 +1,192 @@ +//! Auto-mitigation gate: drives the pure concentration-based `select()` from +//! `blackwall-flow` against two synthetic detections, then executes the +//! chosen mitigation's manager against a real BGP peer (BIRD2): +//! - a concentrated attack (one dominant port) selects FlowSpec, and the +//! `FlowSpecManager` announces a flow-scoped drop rule for it; +//! - a diffuse attack (weight spread across many ports) selects RTBH, and +//! the `RtbhManager` announces a /32 blackhole for it. +//! +//! This proves the whole path — selection, codec, session — end to end +//! against real BIRD, not just the pure `select()` unit tests (Task 2) or +//! the individual FlowSpec/RTBH BIRD gates (C2a/C1b) in isolation. +//! +//! Same RFC 8955 §6 "safe update" validation gotcha as `flowspec_interop.rs` +//! (in `blackwall-bgp/tests`): BIRD only accepts a FlowSpec route whose +//! destination is covered by a unicast route from the same origin AS, so we +//! announce a covering `203.0.113.0/24` route before the FlowSpec rule. +//! BW_BGP_PEER=10.0.0.1:179 cargo test -p blackwall-rtbh --test flowspec_auto_interop -- --ignored --nocapture + +use async_trait::async_trait; +use blackwall_bgp::{spawn, FlowSpecRule, Origin, PeerConfig, Route}; +use blackwall_flow::{ + select, AttackKind, Detection, DetectionEvent, Mitigation, SelectionConfig, Severity, +}; +use blackwall_rtbh::{ + BlackholeJournal, BlackholeOrigin, FlowSpecConfig, FlowSpecController, FlowSpecJournal, + FlowSpecManager, JournalError, RtbhConfig, RtbhController, RtbhManager, +}; +use std::net::IpAddr; +use std::time::Duration; + +/// A no-op RTBH journal: this test only exercises the BGP path against real +/// BIRD, not persistence (covered elsewhere with fakes / real Postgres). +struct NoopBlackholeJournal; + +#[async_trait] +impl BlackholeJournal for NoopBlackholeJournal { + 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(()) + } +} + +/// A no-op FlowSpec journal, for the same reason. +struct NoopFlowSpecJournal; + +#[async_trait] +impl FlowSpecJournal for NoopFlowSpecJournal { + async fn record_announce( + &self, + _rule: FlowSpecRule, + _origin: BlackholeOrigin, + _at_ms: u64, + ) -> Result<(), JournalError> { + Ok(()) + } + async fn record_withdraw(&self, _rule: FlowSpecRule, _at_ms: u64) -> Result<(), JournalError> { + Ok(()) + } +} + +fn concentrated_detection() -> Detection { + Detection { + target: "203.0.113.7".parse().unwrap(), + kind: AttackKind::Volumetric, + observed_pps: 200_000.0, + observed_bps: 2e9, + proto: 17, + top_sources: vec![], + top_ports: vec![(53, 0.95)], + severity: Severity::High, + first_seen_ms: 0, + last_seen_ms: 0, + } +} + +fn diffuse_detection() -> Detection { + Detection { + target: "203.0.113.8".parse().unwrap(), + kind: AttackKind::Volumetric, + observed_pps: 200_000.0, + observed_bps: 2e9, + proto: 17, + top_sources: vec![], + top_ports: 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), + ], + severity: Severity::High, + first_seen_ms: 0, + last_seen_ms: 0, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "needs a live BGP peer (BIRD); run in the netns lab"] +async fn selection_routes_to_flowspec_and_rtbh_on_real_bird() { + let peer: std::net::SocketAddr = std::env::var("BW_BGP_PEER") + .expect("set BW_BGP_PEER=ip:179") + .parse() + .unwrap(); + let (handle, _join) = spawn(PeerConfig { + local_asn: 214_806, + peer_asn: 214_806, + peer_addr: peer, + router_id: "10.222.255.99".parse().unwrap(), + hold_time: 90, + }) + .expect("valid iBGP config"); + tokio::time::sleep(Duration::from_secs(3)).await; // let the session establish + + // Covering unicast route (RFC 8955 §6 "safe update" validation): without + // this, BIRD rejects the FlowSpec rule below as unvalidated. + handle + .announce(Route { + prefix: "203.0.113.0/24".parse().unwrap(), + next_hop: "10.0.0.1".parse().unwrap(), + origin: Origin::Igp, + communities: vec![], + large_communities: vec![], + }) + .await + .expect("announce covering route"); + + let cfg = SelectionConfig { + concentration: 0.8, + max_flows: 4, + rate: 0.0, + }; + + let mut flowspec_mgr = FlowSpecManager::new( + FlowSpecController::new(FlowSpecConfig { + eligible_prefixes: vec!["203.0.113.0/24".parse().unwrap()], + max_rules: 64, + hold_down: Duration::from_secs(0), + max_ttl: None, + }), + handle.clone(), + NoopFlowSpecJournal, + ); + let mut rtbh_mgr = RtbhManager::new( + RtbhController::new(RtbhConfig { + eligible_prefixes: vec!["203.0.113.0/24".parse().unwrap()], + blackhole_communities: vec![(65535, 666)], + next_hop_v4: Some("10.222.255.99".parse().unwrap()), + next_hop_v6: None, + max_blackholes: 64, + hold_down: Duration::from_secs(0), + max_ttl: None, + }), + handle, + NoopBlackholeJournal, + ); + + // Concentrated: a single dominant port (53/udp at 95%) selects FlowSpec — + // a flow-scoped drop rule, leaving the rest of the victim's traffic alone. + let concentrated = concentrated_detection(); + match select(&concentrated, &cfg) { + Mitigation::FlowSpec(rules) => { + flowspec_mgr + .apply_open(concentrated.target, &rules, 0, 1000) + .await; + } + other => panic!("expected FlowSpec for the concentrated detection, got {other:?}"), + } + + // Diffuse: weight spread thin across many ports selects RTBH — no small + // flow set can stop the attack, so the whole victim IP is blackholed. + let diffuse = diffuse_detection(); + match select(&diffuse, &cfg) { + Mitigation::Rtbh => { + rtbh_mgr + .apply_event(&DetectionEvent::Opened(diffuse.clone()), 0, 1000) + .await; + } + other => panic!("expected RTBH for the diffuse detection, got {other:?}"), + } + + tokio::time::sleep(Duration::from_secs(5)).await; // let BIRD import + the scenario assert +}