diff --git a/CHANGELOG.md b/CHANGELOG.md index c5588e3..ccb34f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ All notable changes to this project are documented here, following - Speedtest now runs providers sequentially, measures each download over the full window, and reports the fastest clean provider — fixing wildly variable, under-reporting results on fast links. ### Fixed +- M1 arming hardening (control plane) — closes the three arming-safety follow-ups from the M1 interlock. (**#194**) the boot-time rehydrate path (RTBH/FlowSpec/XDP) that re-announces persisted mitigations on an armed restart no longer strands an entry when the re-announce fails: it now KEEPS the entry and retries on the next tick (a `pending_reapply` self-heal mirroring the journal `pending_mirror`), converging once the session recovers — and, on the mutable-rate planes (FlowSpec/XDP), the retry RE-DERIVES the controller's current rule/action rather than replaying the value captured at queue time, so a stale retry can never revert a fresher successful update; surfaced as `blackwall_{rtbh,flowspec,xdp}_reapply_pending` gauges. (**#193 residuals**) the SIGUSR1 disarm logged its `DISARMED` banner twice (once before the withdrawal, once after) — de-duped to the single accurate post-withdrawal banner; and the disarm withdraw-error tolerance is now tested for FlowSpec + XDP, not just RTBH. + +### Added +- RPKI pre-announce cross-check (deployment #5) — an optional, **fail-open** advisory that warns an operator when blackwall's RTBH blackholes would be dropped by RPKI-validating upstreams. A new `rpki-validator=` directive (absent ⇒ off, non-breaking) points at a Routinator HTTP API; at startup and every `rpki-check-interval` (default 1h) a background task checks each `eligible`/`protected` prefix's **host-length more-specific** (`/32`, `/128` — the shape an RTBH actually announces, because a ROA whose `maxLength` is shorter than the host length makes those blackholes RPKI-INVALID) against `AS`, and WARNs on a state **transition** (not every interval) when a prefix is `invalid`/`not-found` or the validator becomes unreachable. Never blocks detection or arming; an unrecognized/malformed validity response is treated as validator-down (never a spurious `valid`). New pure `blackwall-rpki` crate (classifier/URL/more-specific logic, 100% covered); metrics `blackwall_rpki_validator_up` + `blackwall_rpki_uncovered_prefixes` (present only when the check runs). Enabling it needs Routinator's `http-listen` on the box (its `:8282` is RTR-only); until then the check is simply off. + - `blackwall_flow_pop_last_seen_seconds` reported a nonsense ~epoch-sized value (so any "POP sensor down/stale" alert built on it was dead). Fallout from the monotonic-clock change: the collector stamps `AgentStat.last_seen_ms` from the monotonic clock, but the `/metrics` renderer computed the age against wall-clock epoch — mixing the two domains, the gauge always read ~current-epoch-seconds regardless of how recently the agent was seen. The renderer now uses the collector's monotonic clock (exposed as `blackwall_flow::monotonic_now_ms()`), so a freshly-observed agent reads ~0 s. Found validating the M0 sockpuppet fleet. Detection was never affected (it used the monotonic clock consistently); only the liveness gauge was wrong. - Pre-M0 detection-quality batch (makes the shadow observation's FP/FN signal trustworthy before the AS214806 detection-only deploy; from a full-codebase review + re-triage). **Detector correctness:** windowing/eviction/hold-down now use a **monotonic clock** (`Instant` baseline) so an NTP/wall-clock step can't freeze or force-evict the window; a new **minimum-sample gate** (`--min-samples`, default 8) stops sampling-variance false positives (2 packets at 1-in-65536 no longer extrapolate to a 131k-pps "attack"); the sampling-rate sanity clamp is now **direction-aware** — a low reported rate is clamped up (anti-suppression) while a legitimately-high adaptive-sampling rate is *trusted* up to `expected × --max-sampling-factor` (default 64, floored at 4) instead of being clamped down and hiding a real flood; sFlow decoding is **per-sample resilient** so one malformed sample no longer discards a whole datagram's valid observations (envelope-framing errors stay fatal); and mitigation selection treats destination **port 0** (fragmented / non-TCP-UDP flows) as "no port data" → RTBH, never emitting a useless `dst_port:0` FlowSpec rule (`blackwall-flow`). **Config/API:** the config lexer now treats `#` as a comment only at line-start or after whitespace, so a value like `md5=sec#ret` is no longer silently truncated (`blackwall-config`); `parse_and_resolve()` validates the policy at load for `blackwalld flow` (the M0 entry point) and `bird-config`, so a semantically-invalid config fails at container start instead of mis-behaving through the observation window; and the read API's `?limit=` is clamped to `[1, 1000]` so a negative or huge value can't 500 or dump a whole table (`blackwall-api`). **New observability** (`/metrics`, `blackwall_flow_*`): `min_sample_suppressed_total`, `sampling_near_ceiling_total{pop}`, `sample_decode_errors_total` (distinct from the per-datagram `decode_errors`), and `detections_opened_total` / `detections_cleared_total` — together they let an operator tell a quiet POP from one silently dropping samples. Non-breaking (new CLI flags default to prior behaviour; `--bps-threshold` is documented as L2/frame-length based). - Lab CI reliability (blackwall#88): a stuck in-scenario command could hang a gate silently until the job cap. Root cause was two-fold — the harness ran one-shot commands via an unbounded `Command::output()`, and gate drivers ran `cargo test` *inside* the netns, whose long-lived cargo process (holding the Cargo build lock + a subprocess tree) could wedge in CI where a `sudo`-spawned tree escapes GitHub's step timeout. Fixes: `netns::run` now bounds and kills every command (own process group, drained pipes, reader-join timeouts) so it can never block the caller; daemon teardown reaps with a bounded wait; `spawn_bird` redirects its stdio like the other daemons so a killed lab can't leak a pipe-holder; and gates now run the **pre-built** interop binary directly (`scripts/build-lab-tests.sh` → `target/debug/lab-tests/-`) instead of `cargo test`, removing the cargo lock/daemon from the scenario entirely. diff --git a/Cargo.lock b/Cargo.lock index 48c9aad..711eb71 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,6 +373,17 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "blackwall-rpki" +version = "0.1.0" +dependencies = [ + "ipnet", + "reqwest", + "serde", + "serde_json", + "thiserror 2.0.18", +] + [[package]] name = "blackwall-rtbh" version = "0.1.0" @@ -384,6 +395,7 @@ dependencies = [ "thiserror 2.0.18", "tokio", "tracing", + "tracing-subscriber", ] [[package]] @@ -483,6 +495,7 @@ dependencies = [ "blackwall-flow", "blackwall-metrics", "blackwall-nft", + "blackwall-rpki", "blackwall-rtbh", "blackwall-shaper", "blackwall-speedtest", diff --git a/bin/blackwalld/Cargo.toml b/bin/blackwalld/Cargo.toml index 3adb691..3f1c73f 100644 --- a/bin/blackwalld/Cargo.toml +++ b/bin/blackwalld/Cargo.toml @@ -20,6 +20,7 @@ blackwall-metrics = { path = "../../crates/blackwall-metrics" } blackwall-speedtest = { path = "../../crates/blackwall-speedtest" } blackwall-bgp = { path = "../../crates/blackwall-bgp" } blackwall-rtbh = { path = "../../crates/blackwall-rtbh" } +blackwall-rpki = { path = "../../crates/blackwall-rpki" } blackwall-xdp = { path = "../../crates/blackwall-xdp" } async-trait = { workspace = true } axum = { workspace = true } diff --git a/bin/blackwalld/src/main.rs b/bin/blackwalld/src/main.rs index 57a0c04..17899dc 100644 --- a/bin/blackwalld/src/main.rs +++ b/bin/blackwalld/src/main.rs @@ -682,6 +682,160 @@ fn flowspec_config_from( } } +/// Spawn the periodic RPKI validity cross-check task (#5 C2) when +/// `policy.rpki_validator` is set. +/// +/// Requires an `rtbh` block for `local_asn` (the ASN the check queries the +/// validator with, matching the ASN RTBH announcements actually originate +/// from) — if `rpki_validator` is configured but no `rtbh` block is +/// present, logs one warning and does not spawn (there is nothing sensible +/// to query with). Non-breaking: when `rpki_validator` is unset, this is a +/// no-op, so no config gains a new background task it didn't ask for. +/// +/// Never on the mitigation hot path: this task only fails open (logs a +/// WARN/INFO and updates a metric) — it never touches RTBH/FlowSpec/XDP +/// state itself. +/// +/// Returns `true` iff the task was actually spawned (both `rpki_validator` +/// and `rtbh` present). Callers must use this return value — not just +/// "did I construct the atomics" — to decide whether to wire `Some(...)` +/// into [`metrics::MetricsSources`]'s `rpki_validator_up` / +/// `rpki_uncovered_prefixes` fields: those fields must be `None` (and thus +/// rendered as absent, per the `Option` convention documented on +/// `MetricsSources`) whenever the check isn't actually running, so a daemon +/// with no `rpki-validator=` configured never reports a falsely-healthy +/// `blackwall_rpki_validator_up 1`. +#[must_use] +fn spawn_rpki_check_task( + policy: &blackwall_core::Policy, + validator_up: std::sync::Arc, + uncovered_prefixes: std::sync::Arc, +) -> bool { + let Some(base) = policy.rpki_validator.clone() else { + return false; + }; + let Some(rtbh) = policy.rtbh.as_ref() else { + tracing::warn!( + "rpki-validator is configured but no rtbh block is present; \ + skipping the periodic RPKI validity cross-check (no local ASN \ + to query the validator with)" + ); + return false; + }; + let asn = rtbh.local_asn; + let interval = policy.rpki_check_interval; + + // eligible_prefixes ∪ protected_prefixes: everything an RTBH blackhole + // more-specific could ever be announced for. De-duplicated so a prefix + // listed in both `prefixes` and `protect` isn't checked twice. + let mut seen = std::collections::HashSet::new(); + let prefixes: Vec = policy + .prefixes + .iter() + .chain(policy.protected_prefixes.iter()) + .copied() + .filter(|net| seen.insert(*net)) + .collect(); + + tracing::info!( + base = %base, + asn, + prefixes = prefixes.len(), + interval_secs = interval.as_secs(), + "starting periodic RPKI validity cross-check (#5 C2)" + ); + tokio::spawn(rpki_check_task( + base, + asn, + prefixes, + interval, + validator_up, + uncovered_prefixes, + )); + true +} + +/// The periodic RPKI validity cross-check loop (#5 C2): runs +/// [`blackwall_rpki::check_once`] immediately at startup and then every +/// `interval`, logging a WARN on a fresh validator-down or newly-invalid +/// prefix transition (an INFO on recovery), and updating the +/// `blackwall_rpki_validator_up` / `blackwall_rpki_uncovered_prefixes` +/// gauges. Runs forever — intended to be `tokio::spawn`ed and left detached, +/// like the other supervisory tasks in this daemon. +/// +/// Fails open by construction: every fetch failure is a WARN/metric, never a +/// panic and never a mitigation action — see `blackwall_rpki::fetch_validity`'s +/// docs for the "never a silent pass" contract this task relies on. +async fn rpki_check_task( + base: String, + asn: u32, + prefixes: Vec, + interval: std::time::Duration, + validator_up: std::sync::Arc, + uncovered_prefixes: std::sync::Arc, +) { + use std::sync::atomic::Ordering; + + let client = blackwall_rpki::build_client(); + let mut warn_state = blackwall_rpki::RpkiWarnState::default(); + // Mirrors `warn_state`'s internal validator-reachability bit, kept + // separately here purely so this glue can tell "no change" apart from + // "down→up recovery" for INFO-vs-nothing logging — `RpkiWarnState` + // itself only exposes the WARN-worthy (up→down) transition. + let mut previously_up = true; + // Mirrors `warn_state`'s internal per-prefix map for the same reason: + // telling "no change" apart from "recovered to valid" for INFO logging. + let mut previous_states: std::collections::HashMap = + std::collections::HashMap::new(); + + loop { + let report = blackwall_rpki::check_once(&client, &base, asn, &prefixes).await; + + if warn_state.observe_validator_up(report.validator_up) { + tracing::warn!( + base = %base, + "RPKI validator unreachable; RTBH blackhole RPKI cross-check is \ + failing open (not enforced) until it recovers" + ); + } else if !previously_up && report.validator_up { + tracing::info!(base = %base, "RPKI validator reachable again"); + } + previously_up = report.validator_up; + + let mut uncovered_count = 0usize; + for &(net, state) in &report.per_prefix { + if matches!( + state, + blackwall_rpki::RpkiState::Invalid | blackwall_rpki::RpkiState::NotFound + ) { + uncovered_count += 1; + } + let should_warn = warn_state.observe_prefix(net, state); + let prev = previous_states.insert(net, state); + if should_warn { + tracing::warn!( + prefix = %net, + asn, + state = ?state, + "RTBH blackhole more-specific is RPKI-invalid/uncovered; a \ + validating upstream will silently drop this announcement" + ); + } else if matches!( + prev, + Some(blackwall_rpki::RpkiState::Invalid | blackwall_rpki::RpkiState::NotFound) + ) && state == blackwall_rpki::RpkiState::Valid + { + tracing::info!(prefix = %net, "RTBH blackhole more-specific recovered RPKI validity"); + } + } + + validator_up.store(u8::from(report.validator_up), Ordering::Relaxed); + uncovered_prefixes.store(uncovered_count, Ordering::Relaxed); + + tokio::time::sleep(interval).await; + } +} + /// Construct a host route (`/32` for IPv4, `/128` for IPv6) for `target`. /// /// Local mirror of `blackwall_rtbh`'s crate-private `host_prefix`, used to @@ -831,15 +985,34 @@ async fn bgp_supervisor(mut states: tokio::sync::watch::Receiver, + /// This plane's dedicated apply-failure counter (C2). + apply_failures: Arc, + /// Shared cross-plane rate-cap (C6) skip counters; `None` for XDP, which + /// has no shared rate limiter to report against. + ratecapped: Option>, + /// This plane's dedicated reapply-pending gauge (issue #194 C1). + reapply_pending: Arc, +} + /// Runs until `rx` is closed (i.e. for the process's lifetime, since the /// paired `ChannelSink`'s sender is held by the running collector). async fn rtbh_manager_task( mut manager: blackwall_rtbh::RtbhManager, mut rx: mpsc::Receiver, request_store: std::sync::Arc, - protected_metrics: Arc, - apply_failure_metrics: Arc, - ratecapped_metrics: Arc, + metrics: PlaneMetrics, mut disarm_rx: tokio::sync::broadcast::Receiver<()>, ) where B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, @@ -871,7 +1044,6 @@ async fn rtbh_manager_task( // is treated the same as `Ok`. match disarmed { Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - tracing::warn!("RTBH: DISARMED — mitigations withdrawn, now recording only"); manager.disarm(mono_now()).await; } Err(tokio::sync::broadcast::error::RecvError::Closed) => { @@ -884,16 +1056,21 @@ async fn rtbh_manager_task( // elapses is deferred and never completed. manager.tick(mono_now(), wall_now()).await; - protected_metrics.rtbh.store( + metrics.protected.rtbh.store( manager.protected_skipped(), std::sync::atomic::Ordering::Relaxed, ); - apply_failure_metrics.store( + metrics.apply_failures.store( manager.apply_failures(), std::sync::atomic::Ordering::Relaxed, ); - ratecapped_metrics.rtbh.store( - manager.ratecapped(), + if let Some(ratecapped) = &metrics.ratecapped { + ratecapped + .rtbh + .store(manager.ratecapped(), std::sync::atomic::Ordering::Relaxed); + } + metrics.reapply_pending.store( + manager.reapply_pending(), std::sync::atomic::Ordering::Relaxed, ); @@ -992,9 +1169,7 @@ async fn flowspec_manager_task( mut manager: blackwall_rtbh::FlowSpecManager, mut rx: mpsc::Receiver, request_store: std::sync::Arc, - protected_metrics: Arc, - apply_failure_metrics: Arc, - ratecapped_metrics: Arc, + metrics: PlaneMetrics, mut disarm_rx: tokio::sync::broadcast::Receiver<()>, ) where B: blackwall_rtbh::manager::BgpExecutor + Send + 'static, @@ -1028,7 +1203,6 @@ async fn flowspec_manager_task( // `Lagged` delivery still counts as "disarm was requested". match disarmed { Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - tracing::warn!("FlowSpec: DISARMED — mitigations withdrawn, now recording only"); manager.disarm(mono_now()).await; } Err(tokio::sync::broadcast::error::RecvError::Closed) => { @@ -1040,16 +1214,22 @@ async fn flowspec_manager_task( // Mandatory: completes deferred clears / TTL expiry. manager.tick(mono_now(), wall_now()).await; - protected_metrics.flowspec.store( + metrics.protected.flowspec.store( manager.protected_skipped(), std::sync::atomic::Ordering::Relaxed, ); - apply_failure_metrics.store( + metrics.apply_failures.store( manager.apply_failures(), std::sync::atomic::Ordering::Relaxed, ); - ratecapped_metrics.flowspec.store( - manager.ratecapped(), + if let Some(ratecapped) = &metrics.ratecapped { + ratecapped.flowspec.store( + manager.ratecapped(), + std::sync::atomic::Ordering::Relaxed, + ); + } + metrics.reapply_pending.store( + manager.reapply_pending(), std::sync::atomic::Ordering::Relaxed, ); @@ -1210,8 +1390,7 @@ async fn xdp_manager_task( mut rx: mpsc::Receiver, request_store: std::sync::Arc, auto_enabled: bool, - protected_metrics: Arc, - apply_failure_metrics: Arc, + metrics: PlaneMetrics, mut disarm_rx: tokio::sync::broadcast::Receiver<()>, ) where J: blackwall_xdp::XdpJournal + 'static, @@ -1239,7 +1418,6 @@ async fn xdp_manager_task( // `Lagged` delivery still counts as "disarm was requested". match disarmed { Ok(()) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { - tracing::warn!("XDP: DISARMED — mitigations withdrawn, now recording only"); manager.disarm(mono_now()).await; } Err(tokio::sync::broadcast::error::RecvError::Closed) => { @@ -1251,14 +1429,18 @@ async fn xdp_manager_task( // Drains any journal mirror-writes queued by a transient DB blip. manager.tick().await; - protected_metrics.xdp.store( + metrics.protected.xdp.store( manager.protected_skipped(), std::sync::atomic::Ordering::Relaxed, ); - apply_failure_metrics.store( + metrics.apply_failures.store( manager.apply_failures(), std::sync::atomic::Ordering::Relaxed, ); + metrics.reapply_pending.store( + manager.reapply_pending(), + std::sync::atomic::Ordering::Relaxed, + ); match request_store.xdp_pending_requests().await { Ok(reqs) => { @@ -1587,6 +1769,50 @@ async fn run() -> Result<(), Box> { // unconditionally — harmless all-zero counters when no `rtbh` // block is configured or `max-new-per-min` is unset. let ratecapped_metrics = std::sync::Arc::new(shadow::RatecappedMetrics::default()); + // `rehydrate` re-announces queued for a self-heal retry after a + // failed BGP announce on restart (issue #194): copied from + // `RtbhManager::reapply_pending` on every tick, mirroring + // `rtbh_apply_failure_metrics` above. Built unconditionally — + // harmless all-zero gauge when no `rtbh` block is configured. + let rtbh_reapply_pending_metrics = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + // `rehydrate` re-announces queued for a self-heal retry after a + // failed BGP announce on restart (issue #194): copied from + // `FlowSpecManager::reapply_pending` on every tick, mirroring + // `rtbh_reapply_pending_metrics` above. Built unconditionally — + // harmless all-zero gauge when no `flowspec` block is configured. + let flowspec_reapply_pending_metrics = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + // `reapply_active` re-applies queued for a self-heal retry after + // a failed executor apply on restart (issue #194): copied from + // `blackwall_xdp::manager::XdpManager::reapply_pending` on every + // tick, mirroring `rtbh_reapply_pending_metrics` above. Built + // unconditionally — harmless all-zero gauge when no `xdp` block + // is configured. + let xdp_reapply_pending_metrics = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + // Bundled per-plane `/metrics` argument for the three + // `*_manager_task`s (prep refactor for #194 C1 — see + // `PlaneMetrics`'s doc comment). XDP has no shared rate limiter, + // so its `ratecapped` is `None`. + let rtbh_plane_metrics = PlaneMetrics { + protected: protected_skipped_metrics.clone(), + apply_failures: rtbh_apply_failure_metrics.clone(), + ratecapped: Some(ratecapped_metrics.clone()), + reapply_pending: rtbh_reapply_pending_metrics.clone(), + }; + let flowspec_plane_metrics = PlaneMetrics { + protected: protected_skipped_metrics.clone(), + apply_failures: flowspec_apply_failure_metrics.clone(), + ratecapped: Some(ratecapped_metrics.clone()), + reapply_pending: flowspec_reapply_pending_metrics.clone(), + }; + let xdp_plane_metrics = PlaneMetrics { + protected: protected_skipped_metrics.clone(), + apply_failures: xdp_apply_failure_metrics.clone(), + ratecapped: None, + reapply_pending: xdp_reapply_pending_metrics.clone(), + }; // In-daemon disarm kill switch (C5): `blackwall_armed` starts at // 1 (live) or 0 (shadow) and is flipped to 0 exactly once, on a // SIGUSR1 disarm — there is no path back to 1 short of a @@ -1599,6 +1825,23 @@ async fn run() -> Result<(), Box> { std::sync::Arc::new(std::sync::atomic::AtomicU8::new(u8::from(!policy.shadow))); let (disarm_tx, _disarm_rx) = tokio::sync::broadcast::channel::<()>(8); + // Periodic RPKI validity cross-check (#5 C2): a no-op unless + // `rpki-validator=` is configured (and an `rtbh` block is + // present — see `spawn_rpki_check_task`). The atomics are + // constructed unconditionally (harmless — starts at + // "reachable"/"nothing uncovered"), but `rpki_check_running` + // (the task's own condition) gates whether they are actually + // wired into `/metrics` below: the metric must be *absent*, + // not just idle, when the check never runs. + let rpki_validator_up = std::sync::Arc::new(std::sync::atomic::AtomicU8::new(1)); + let rpki_uncovered_prefixes = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let rpki_check_running = spawn_rpki_check_task( + &policy, + rpki_validator_up.clone(), + rpki_uncovered_prefixes.clone(), + ); + let sink: std::sync::Arc = match policy.rtbh.clone() { None => std::sync::Arc::new(blackwall_state::PgMitigationSink::new(store.clone())), @@ -1653,9 +1896,7 @@ async fn run() -> Result<(), Box> { manager, rx, store.clone(), - protected_skipped_metrics.clone(), - rtbh_apply_failure_metrics.clone(), - ratecapped_metrics.clone(), + rtbh_plane_metrics.clone(), disarm_tx.subscribe(), )); None @@ -1707,9 +1948,7 @@ async fn run() -> Result<(), Box> { manager, rx, store.clone(), - protected_skipped_metrics.clone(), - rtbh_apply_failure_metrics.clone(), - ratecapped_metrics.clone(), + rtbh_plane_metrics.clone(), disarm_tx.subscribe(), )); Some(bgp) @@ -1758,9 +1997,7 @@ async fn run() -> Result<(), Box> { fs_manager, fs_rx, store.clone(), - protected_skipped_metrics.clone(), - flowspec_apply_failure_metrics.clone(), - ratecapped_metrics.clone(), + flowspec_plane_metrics.clone(), disarm_tx.subscribe(), )); } @@ -1810,9 +2047,7 @@ async fn run() -> Result<(), Box> { fs_manager, fs_rx, store.clone(), - protected_skipped_metrics.clone(), - flowspec_apply_failure_metrics.clone(), - ratecapped_metrics.clone(), + flowspec_plane_metrics.clone(), disarm_tx.subscribe(), )); } @@ -2002,8 +2237,7 @@ async fn run() -> Result<(), Box> { xdp_rx, store.clone(), auto_enabled, - protected_skipped_metrics.clone(), - xdp_apply_failure_metrics.clone(), + xdp_plane_metrics.clone(), disarm_tx.subscribe(), )) } else { @@ -2029,8 +2263,7 @@ async fn run() -> Result<(), Box> { xdp_rx, store.clone(), auto_enabled, - protected_skipped_metrics.clone(), - xdp_apply_failure_metrics.clone(), + xdp_plane_metrics.clone(), disarm_tx.subscribe(), )) }; @@ -2068,7 +2301,13 @@ async fn run() -> Result<(), Box> { rtbh_apply_failures: Some(rtbh_apply_failure_metrics.clone()), flowspec_apply_failures: Some(flowspec_apply_failure_metrics.clone()), xdp_apply_failures: Some(xdp_apply_failure_metrics.clone()), + rtbh_reapply_pending: Some(rtbh_reapply_pending_metrics.clone()), + flowspec_reapply_pending: Some(flowspec_reapply_pending_metrics.clone()), + xdp_reapply_pending: Some(xdp_reapply_pending_metrics.clone()), armed: Some(blackwall_armed.clone()), + rpki_validator_up: rpki_check_running.then(|| rpki_validator_up.clone()), + rpki_uncovered_prefixes: rpki_check_running + .then(|| rpki_uncovered_prefixes.clone()), }; tokio::spawn(metrics::metrics_server(metrics_listen, sources)); } @@ -2156,6 +2395,22 @@ async fn run() -> Result<(), Box> { ensure_interface_exists(&policy.interface)?; ensure_flowtable_devices_exist(&policy)?; + // Periodic RPKI validity cross-check (#5 C2): a no-op unless + // `rpki-validator=` is configured (and an `rtbh` block is + // present — see `spawn_rpki_check_task`). The deception engine + // (`run`) does not itself manage RTBH/FlowSpec/XDP, but this + // monitoring task is independent of that and belongs wherever + // the policy is parsed. `rpki_check_running` gates whether the + // atomics are wired into `/metrics` below. + let rpki_validator_up = std::sync::Arc::new(std::sync::atomic::AtomicU8::new(1)); + let rpki_uncovered_prefixes = + std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let rpki_check_running = spawn_rpki_check_task( + &policy, + rpki_validator_up.clone(), + rpki_uncovered_prefixes.clone(), + ); + // Connect and migrate the store early so discovery can persist its results. let store = blackwall_state::Store::connect(&database_url).await?; store.migrate().await?; @@ -2327,7 +2582,13 @@ async fn run() -> Result<(), Box> { rtbh_apply_failures: None, flowspec_apply_failures: None, xdp_apply_failures: None, + rtbh_reapply_pending: None, + flowspec_reapply_pending: None, + xdp_reapply_pending: None, armed: None, + rpki_validator_up: rpki_check_running.then(|| rpki_validator_up.clone()), + rpki_uncovered_prefixes: rpki_check_running + .then(|| rpki_uncovered_prefixes.clone()), }; tokio::spawn(metrics::metrics_server(metrics_listen, sources)); } diff --git a/bin/blackwalld/src/metrics.rs b/bin/blackwalld/src/metrics.rs index 66ebebb..58b0d2d 100644 --- a/bin/blackwalld/src/metrics.rs +++ b/bin/blackwalld/src/metrics.rs @@ -55,6 +55,24 @@ pub(crate) struct MetricsSources { /// `xdp` block is configured. Copied from the manager once per tick, /// mirroring `rtbh_apply_failures`. pub xdp_apply_failures: Option>, + /// `rehydrate` re-announces currently queued for a self-heal retry after + /// a failed BGP announce on restart (issue #194, + /// `RtbhManager::reapply_pending`); `None` when no `rtbh` block is + /// configured. Copied from the manager once per tick, mirroring how + /// `rtbh_apply_failures` reaches this endpoint. + pub rtbh_reapply_pending: Option>, + /// `rehydrate` re-announces currently queued for a self-heal retry after + /// a failed BGP announce on restart (issue #194, + /// `FlowSpecManager::reapply_pending`); `None` when no `flowspec` block + /// is configured. Copied from the manager once per tick, mirroring + /// `rtbh_reapply_pending`. + pub flowspec_reapply_pending: Option>, + /// `reapply_active` re-applies currently queued for a self-heal retry + /// after a failed executor apply on restart (issue #194, + /// `blackwall_xdp::manager::XdpManager::reapply_pending`); `None` when no + /// `xdp` block is configured. Copied from the manager once per tick, + /// mirroring `rtbh_reapply_pending`. + pub xdp_reapply_pending: Option>, /// Per-plane cross-plane new-mitigation rate cap (C6) skip counters /// (`RtbhManager`/`FlowSpecManager::ratecapped`); `None` outside the flow /// daemon (no managers to cap). Populated (all-zero) even when no `rtbh` @@ -67,6 +85,19 @@ pub(crate) struct MetricsSources { /// path back to `1` short of a restart. `None` outside the flow daemon /// (no RTBH/FlowSpec/XDP managers to arm). pub armed: Option>, + /// Whether the RPKI validator (`rpki-validator=`) answered the most + /// recent periodic validity check pass: `1` reachable, `0` unreachable + /// (fail-open, C2). `None` when `rpki-validator` is not configured, or + /// it is configured but no `rtbh` block is present (no ASN to query + /// with — the task never spawns). Built unconditionally as `Some(1)` + /// (assumed reachable) whenever the check task is spawned, mirroring + /// `armed`'s "flipped on observation" pattern. + pub rpki_validator_up: Option>, + /// Count of RTBH-eligible prefixes whose RPKI validity state was + /// `invalid` or `not-found` at the last periodic check pass (C2) — i.e. + /// blackhole more-specifics a validating upstream would silently drop. + /// `None` under the same conditions as `rpki_validator_up`. + pub rpki_uncovered_prefixes: Option>, } /// Correctly-rounded `u64 -> f64` without an `as` cast: `u32 -> f64` is exact @@ -171,6 +202,32 @@ async fn gather(sources: &MetricsSources) -> Vec { value: u64_to_f64(xdp_apply_failures.load(std::sync::atomic::Ordering::Relaxed)), }); } + if let Some(rtbh_reapply_pending) = &sources.rtbh_reapply_pending { + m.push(Metric { + name: "blackwall_rtbh_reapply_pending", + help: "RTBH rehydrate re-announces queued for a self-heal retry after a failed BGP announce on restart (#194)", + kind: MetricKind::Gauge, + value: count_to_f64(rtbh_reapply_pending.load(std::sync::atomic::Ordering::Relaxed)), + }); + } + if let Some(flowspec_reapply_pending) = &sources.flowspec_reapply_pending { + m.push(Metric { + name: "blackwall_flowspec_reapply_pending", + help: "FlowSpec rehydrate re-announces queued for a self-heal retry after a failed BGP announce on restart (#194)", + kind: MetricKind::Gauge, + value: count_to_f64( + flowspec_reapply_pending.load(std::sync::atomic::Ordering::Relaxed), + ), + }); + } + if let Some(xdp_reapply_pending) = &sources.xdp_reapply_pending { + m.push(Metric { + name: "blackwall_xdp_reapply_pending", + help: "XDP reapply_active re-applies queued for a self-heal retry after a failed executor apply on restart (#194)", + kind: MetricKind::Gauge, + value: count_to_f64(xdp_reapply_pending.load(std::sync::atomic::Ordering::Relaxed)), + }); + } if let Some(armed) = &sources.armed { m.push(Metric { name: "blackwall_armed", @@ -179,6 +236,22 @@ async fn gather(sources: &MetricsSources) -> Vec { value: f64::from(armed.load(std::sync::atomic::Ordering::Relaxed)), }); } + if let Some(validator_up) = &sources.rpki_validator_up { + m.push(Metric { + name: "blackwall_rpki_validator_up", + help: "Whether the RPKI validator answered the last periodic validity check: 1 reachable, 0 unreachable (fail-open, C2)", + kind: MetricKind::Gauge, + value: f64::from(validator_up.load(std::sync::atomic::Ordering::Relaxed)), + }); + } + if let Some(uncovered) = &sources.rpki_uncovered_prefixes { + m.push(Metric { + name: "blackwall_rpki_uncovered_prefixes", + help: "RTBH-eligible prefixes whose RPKI validity was invalid/not-found at the last periodic check (C2)", + kind: MetricKind::Gauge, + value: count_to_f64(uncovered.load(std::sync::atomic::Ordering::Relaxed)), + }); + } let s = &sources.store; match s.list_active_blackholes().await { diff --git a/crates/blackwall-config/src/parser.rs b/crates/blackwall-config/src/parser.rs index c17beb2..e0c3f4a 100644 --- a/crates/blackwall-config/src/parser.rs +++ b/crates/blackwall-config/src/parser.rs @@ -30,6 +30,8 @@ pub fn parse(lines: &[Line]) -> Result { let mut pops: Vec = Vec::new(); let mut shadow = false; let mut protected_prefixes: Vec = Vec::new(); + let mut rpki_validator: Option = None; + let mut rpki_check_interval = std::time::Duration::from_secs(3600); let mut i = 0; while i < lines.len() { @@ -805,6 +807,38 @@ pub fn parse(lines: &[Line]) -> Result { } shadow = true; } + // `rpki-validator=` / `rpki-check-interval=` are single + // `key=value` tokens (the whole line), not a leading keyword + // followed by `key=value` pairs like `metrics`/`rtbh` — matched + // by prefix rather than as an exact `directive` string. + d if d.starts_with("rpki-validator=") => { + expect_len(line, 1, "rpki-validator=")?; + if rpki_validator.is_some() { + return Err(ConfigError::BadValue { + line: line.number, + what: "rpki-validator", + value: "duplicate".to_owned(), + }); + } + let url = d + .strip_prefix("rpki-validator=") + .expect("checked by starts_with guard above"); + if url.is_empty() { + return Err(ConfigError::BadValue { + line: line.number, + what: "rpki-validator", + value: "empty url".to_owned(), + }); + } + rpki_validator = Some(url.to_owned()); + } + d if d.starts_with("rpki-check-interval=") => { + expect_len(line, 1, "rpki-check-interval=")?; + let v = d + .strip_prefix("rpki-check-interval=") + .expect("checked by starts_with guard above"); + rpki_check_interval = parse_duration(line, v)?; + } other => { return Err(ConfigError::UnknownDirective { line: line.number, @@ -851,6 +885,8 @@ pub fn parse(lines: &[Line]) -> Result { stateless_tcp_ports, shadow, protected_prefixes, + rpki_validator, + rpki_check_interval, }) } @@ -2326,4 +2362,85 @@ flowspec concentration=0.8 max-flows=4 rate=0 max-rules=256 hold-down=60s bogus= let p = parse_text("interface wan eth0\nipv4 203.0.113.0/24\n").unwrap(); assert!(p.protected_prefixes.is_empty()); } + + #[test] + fn parses_rpki_validator() { + let p = parse_text("interface wan eth0\nrpki-validator=http://h:8323\n").unwrap(); + assert_eq!(p.rpki_validator, Some("http://h:8323".to_owned())); + } + + #[test] + fn rpki_validator_absent_by_default() { + let p = parse_text("interface wan eth0\n").unwrap(); + assert_eq!(p.rpki_validator, None); + } + + #[test] + fn parses_rpki_check_interval() { + let p = parse_text( + "interface wan eth0\nrpki-validator=http://h:8323\nrpki-check-interval=30m\n", + ) + .unwrap(); + assert_eq!(p.rpki_check_interval, std::time::Duration::from_secs(1800)); + } + + #[test] + fn rpki_check_interval_defaults_to_one_hour() { + let p = parse_text("interface wan eth0\nrpki-validator=http://h:8323\n").unwrap(); + assert_eq!(p.rpki_check_interval, std::time::Duration::from_secs(3600)); + } + + #[test] + fn rpki_check_interval_defaults_even_without_validator() { + let p = parse_text("interface wan eth0\n").unwrap(); + assert_eq!(p.rpki_check_interval, std::time::Duration::from_secs(3600)); + } + + #[test] + fn rejects_duplicate_rpki_validator() { + let err = parse_text( + "interface wan eth0\nrpki-validator=http://a:1\nrpki-validator=http://b:2\n", + ) + .unwrap_err(); + assert!( + matches!( + err, + ConfigError::BadValue { + what: "rpki-validator", + .. + } + ), + "got {err:?}" + ); + } + + #[test] + fn rejects_empty_rpki_validator_url() { + let err = parse_text("interface wan eth0\nrpki-validator=\n").unwrap_err(); + assert!( + matches!( + err, + ConfigError::BadValue { + what: "rpki-validator", + .. + } + ), + "got {err:?}" + ); + } + + #[test] + fn rejects_bad_rpki_check_interval() { + let err = parse_text("interface wan eth0\nrpki-check-interval=notaduration\n").unwrap_err(); + assert!( + matches!( + err, + ConfigError::BadValue { + what: "duration", + .. + } + ), + "got {err:?}" + ); + } } diff --git a/crates/blackwall-core/src/policy.rs b/crates/blackwall-core/src/policy.rs index 8d6246c..f656ed8 100644 --- a/crates/blackwall-core/src/policy.rs +++ b/crates/blackwall-core/src/policy.rs @@ -81,4 +81,14 @@ pub struct Policy { /// similar always-safe destinations), set via the repeatable `protect` /// directive. Empty (the default) protects nothing extra. pub protected_prefixes: Vec, + /// Base URL of a Routinator `/api/v1/validity` RPKI validator (the + /// `rpki-validator=` directive, e.g. `http://h:8323`), used to + /// cross-check that RTBH blackhole more-specifics will not be + /// RPKI-invalid at validating upstreams. `None` (the default) disables + /// the check entirely. + pub rpki_validator: Option, + /// How often to re-run the RPKI cross-check (the `rpki-check-interval=` + /// directive). Defaults to one hour; meaningless when `rpki_validator` + /// is `None`. + pub rpki_check_interval: std::time::Duration, } diff --git a/crates/blackwall-core/src/resolve.rs b/crates/blackwall-core/src/resolve.rs index 88956d1..cf6a4c3 100644 --- a/crates/blackwall-core/src/resolve.rs +++ b/crates/blackwall-core/src/resolve.rs @@ -141,6 +141,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } @@ -308,6 +310,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; let resolved = policy.resolve().expect("empty policy resolves"); assert!(resolved.is_empty()); diff --git a/crates/blackwall-deception/tests/interop.rs b/crates/blackwall-deception/tests/interop.rs index 831d04e..90ef9a9 100644 --- a/crates/blackwall-deception/tests/interop.rs +++ b/crates/blackwall-deception/tests/interop.rs @@ -125,6 +125,8 @@ async fn serves_deception_banner() { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000 @@ -193,6 +195,8 @@ async fn serves_deception_under_load() { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; // Apply the REAL nft ruleset: deception TCP on the prefix -> tproxy :61000 @@ -283,6 +287,8 @@ fn serves_stateless_syn_cookie() { stateless_tcp_ports: vec![8080], protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; // Apply the REAL nft ruleset: stateless-tcp TCP on 8080 -> nfqueue @@ -369,6 +375,8 @@ fn serves_stateless_syn_cookie_v6() { stateless_tcp_ports: vec![8080], protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; blackwall_nft::apply(&policy).expect("nft apply"); diff --git a/crates/blackwall-discovery/src/reconcile.rs b/crates/blackwall-discovery/src/reconcile.rs index cce7d04..bb0a087 100644 --- a/crates/blackwall-discovery/src/reconcile.rs +++ b/crates/blackwall-discovery/src/reconcile.rs @@ -133,6 +133,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } diff --git a/crates/blackwall-nft/src/render.rs b/crates/blackwall-nft/src/render.rs index 17ad0e6..1ff4605 100644 --- a/crates/blackwall-nft/src/render.rs +++ b/crates/blackwall-nft/src/render.rs @@ -724,6 +724,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } @@ -756,6 +758,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } @@ -1264,6 +1268,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; let ruleset = render(&policy).expect("render empty"); // No resolved services, so real_v4 and real_v6 sets are empty. diff --git a/crates/blackwall-nft/tests/apply_netns.rs b/crates/blackwall-nft/tests/apply_netns.rs index 46206a1..30a9b9a 100644 --- a/crates/blackwall-nft/tests/apply_netns.rs +++ b/crates/blackwall-nft/tests/apply_netns.rs @@ -37,6 +37,8 @@ fn sample() -> Policy { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } @@ -96,6 +98,8 @@ fn stale_set_elements_removed_on_second_apply() { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), }; blackwall_nft::apply(&policy_empty).expect("second apply"); diff --git a/crates/blackwall-rpki/Cargo.toml b/crates/blackwall-rpki/Cargo.toml new file mode 100644 index 0000000..235eeb4 --- /dev/null +++ b/crates/blackwall-rpki/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "blackwall-rpki" +version = "0.1.0" +edition.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +ipnet = { workspace = true } +reqwest = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +thiserror = { workspace = true } + +[lints] +workspace = true diff --git a/crates/blackwall-rpki/src/fetch.rs b/crates/blackwall-rpki/src/fetch.rs new file mode 100644 index 0000000..3946e52 --- /dev/null +++ b/crates/blackwall-rpki/src/fetch.rs @@ -0,0 +1,91 @@ +//! The HTTP fetch against a live Routinator `/api/v1/validity` endpoint, and +//! the per-check-pass driver that runs it over a set of prefixes. +//! +//! This is the crate's one carve-out from its "no I/O" invariant (see the +//! crate-level docs) — kept in its own file, coverage-excluded, so the rest +//! of the crate stays trivially unit testable. `aggregate_report` (the pure +//! half of [`check_once`]) lives in `lib.rs` and *is* unit tested. + +use std::time::Duration; + +use crate::{aggregate_report, classify, host_more_specific, validity_url, RpkiParseError}; +use crate::{RpkiReport, RpkiState}; + +/// A request/HTTP/timeout/parse failure while checking one prefix against +/// the RPKI validator. Any variant means "treat this prefix (and, per +/// [`aggregate_report`]'s rule, possibly the whole validator) as down" — the +/// caller must fail **open**, never panic, never silently treat it as valid. +#[derive(Debug, thiserror::Error)] +pub enum FetchError { + /// The request itself failed: DNS, connect, TLS, the 5s timeout, or a + /// non-2xx HTTP status. + #[error("RPKI validator request failed: {0}")] + Request(#[from] reqwest::Error), + /// The request succeeded but the response body did not parse into a + /// recognized [`RpkiState`] (see [`classify`]). + #[error("RPKI validator response could not be parsed: {0}")] + Parse(#[from] RpkiParseError), +} + +/// The per-request timeout for a single validity check. Short and fixed — +/// this runs on a periodic background task, never the mitigation hot path, +/// but a hung validator must not stall the check pass indefinitely. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); + +/// Build a default [`reqwest::Client`] for the periodic RPKI check task. +/// +/// A thin re-export so callers (`blackwalld`'s periodic task) never need +/// `reqwest` as a direct dependency themselves — it stays confined to this +/// crate's coverage-excluded I/O boundary. Build once at task startup and +/// reuse it across every [`check_once`] pass (connection reuse). +#[must_use] +pub fn build_client() -> reqwest::Client { + reqwest::Client::new() +} + +/// `GET url` and classify the response body via [`classify`]. +/// +/// Any request/HTTP/timeout error, or a `classify` parse error, is returned +/// as [`FetchError`] — never a panic, never a silent pass. The caller (the +/// periodic checker in `blackwalld`) maps this to `blackwall_rpki_validator_up 0`. +/// +/// # Errors +/// +/// Returns [`FetchError::Request`] on any transport/HTTP-status/timeout +/// failure, or [`FetchError::Parse`] if the response body did not parse into +/// a recognized [`RpkiState`]. +pub async fn fetch_validity(client: &reqwest::Client, url: &str) -> Result { + let response = client + .get(url) + .timeout(REQUEST_TIMEOUT) + .send() + .await? + .error_for_status()?; + let body = response.text().await?; + Ok(classify(&body)?) +} + +/// Run one RPKI validity check pass: form the host more-specific of each of +/// `prefixes`, fetch+classify each against `base` (the validator's base URL) +/// for `asn` (the querying/announcing ASN, `RtbhPolicy.local_asn`), and +/// aggregate the results (see [`aggregate_report`] for the validator-up +/// rule). +/// +/// Never panics: every fetch failure becomes an omitted `per_prefix` entry +/// and, per [`aggregate_report`], may flip `validator_up` to `false` — this +/// function fails open, it never blocks a mitigation. +pub async fn check_once( + client: &reqwest::Client, + base: &str, + asn: u32, + prefixes: &[ipnet::IpNet], +) -> RpkiReport { + let mut results = Vec::with_capacity(prefixes.len()); + for net in prefixes { + let ms = host_more_specific(net); + let url = validity_url(base, asn, &ms); + let outcome = fetch_validity(client, &url).await; + results.push((ms, outcome)); + } + aggregate_report(results) +} diff --git a/crates/blackwall-rpki/src/lib.rs b/crates/blackwall-rpki/src/lib.rs new file mode 100644 index 0000000..98d138d --- /dev/null +++ b/crates/blackwall-rpki/src/lib.rs @@ -0,0 +1,414 @@ +//! Pure RPKI pre-announce cross-check logic for RTBH blackholes. +//! +//! At M1 (arming) an RTBH mitigation announces a `/32`/`/128` +//! "more-specific" blackhole route. If the covering ROA's `maxLength` is +//! shorter than that (a common, legitimate anti-deaggregation ROA), the +//! more-specific is RPKI-**invalid** and validating upstreams silently drop +//! it — the blackhole never takes effect. This crate holds the pure, +//! I/O-free pieces of the cross-check against a +//! [Routinator](https://nlnetlabs.nl/projects/routinator/) 0.14.2 +//! `/api/v1/validity` endpoint: the more-specific former, the response +//! classifier, and the request-URL builder. The periodic *task* (the tokio +//! loop that drives the checks on an interval) lives in `blackwalld`, not +//! here. +//! +//! The one deliberate exception to "no I/O" is the [`fetch`] module: it +//! owns the actual `reqwest` HTTP call to the validator and is the sole, +//! isolated I/O boundary of this crate — kept intentionally small, +//! coverage-excluded, and free of any classification/formation logic of its +//! own. Everything else in this crate (the classifier, the more-specific +//! former, the URL builder, [`aggregate_report`], [`RpkiWarnState`]) is pure +//! and trivially unit-testable; that separation is the design invariant to +//! keep, not the absence of a `fetch` module. + +use std::collections::HashMap; + +use serde::Deserialize; + +mod fetch; +pub use fetch::{build_client, check_once, fetch_validity, FetchError}; + +/// The RPKI validity state of a single announced prefix, as classified from +/// a Routinator `/api/v1/validity` response. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RpkiState { + /// A covering ROA authorizes this exact origin ASN and prefix length. + Valid, + /// A covering ROA exists but does not authorize this origin ASN and/or + /// prefix length (e.g. the ROA's `maxLength` is shorter than `/32`). + /// Validating upstreams will drop an announcement in this state. + Invalid, + /// No covering ROA exists for this prefix at all. + NotFound, +} + +/// The Routinator response could not be parsed into a recognized +/// [`RpkiState`]. +/// +/// This covers malformed JSON, a missing `validated_route.validity.state` +/// path, and any `state` string outside the pinned Routinator 0.14.2 +/// contract (`"valid"`/`"invalid"`/`"not-found"`) — including RIPE's +/// `"unknown"` and any future schema drift. Callers must treat this as +/// "validator down" and fail **open** (skip the check, don't silently treat +/// it as valid), never as a silent pass. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("could not parse RPKI validity response")] +pub struct RpkiParseError; + +/// Deserialization shape for a Routinator 0.14.2 `/api/v1/validity` +/// response. Only the fields the classifier needs are modeled; the response +/// carries more (`route.origin_asn`, `route.prefix`, `reason`, ...). +#[derive(Debug, Deserialize)] +struct ValidityResponse { + validated_route: Option, +} + +#[derive(Debug, Deserialize)] +struct ValidatedRoute { + validity: Validity, +} + +#[derive(Debug, Deserialize)] +struct Validity { + state: String, +} + +/// Classify a Routinator `/api/v1/validity` JSON response body. +/// +/// Reads `validated_route.validity.state` and maps `"valid"` → [`RpkiState::Valid`], +/// `"invalid"` → [`RpkiState::Invalid`], `"not-found"` → [`RpkiState::NotFound`]. +/// Any other value, or a missing/malformed `validated_route`/`validity`/`state` +/// path (including RIPE's `"unknown"` or a future schema change), returns +/// [`RpkiParseError`] so the caller fails open. +/// +/// # Errors +/// +/// Returns [`RpkiParseError`] if `json` is not valid JSON, does not contain +/// the expected `validated_route.validity.state` path, or `state` is not +/// one of the three recognized values. +pub fn classify(json: &str) -> Result { + let response: ValidityResponse = serde_json::from_str(json).map_err(|_| RpkiParseError)?; + let validated_route = response.validated_route.ok_or(RpkiParseError)?; + match validated_route.validity.state.as_str() { + "valid" => Ok(RpkiState::Valid), + "invalid" => Ok(RpkiState::Invalid), + "not-found" => Ok(RpkiState::NotFound), + _ => Err(RpkiParseError), + } +} + +/// Form the "host more-specific" of `net`: the network address at host +/// prefix length (`/32` for IPv4, `/128` for IPv6). This is the exact route +/// an RTBH mitigation announces, and the one that must be checked against +/// RPKI — a covering ROA that authorizes the wider `net` does not +/// necessarily authorize this narrower announcement (a `maxLength` shorter +/// than the host length makes it RPKI-invalid). +pub fn host_more_specific(net: &ipnet::IpNet) -> ipnet::IpNet { + match net { + ipnet::IpNet::V4(v4) => ipnet::IpNet::V4( + ipnet::Ipv4Net::new(v4.network(), 32).expect("32 is a valid IPv4 prefix length"), + ), + ipnet::IpNet::V6(v6) => ipnet::IpNet::V6( + ipnet::Ipv6Net::new(v6.network(), 128).expect("128 is a valid IPv6 prefix length"), + ), + } +} + +/// Build the Routinator `/api/v1/validity` request URL for `ms` announced +/// from `asn`. +/// +/// `base` is the validator's base URL with no trailing slash (e.g. +/// `http://h:8323`, from the `rpki-validator=` config directive). `ms` is +/// typically the output of [`host_more_specific`]. +pub fn validity_url(base: &str, asn: u32, ms: &ipnet::IpNet) -> String { + format!("{base}/api/v1/validity/AS{asn}/{ms}") +} + +/// The aggregate result of one [`fetch::check_once`] pass over a set of +/// RTBH-eligible prefixes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RpkiReport { + /// Whether the validator itself was reachable during this pass. See + /// [`aggregate_report`] for exactly how this is derived from the + /// per-prefix fetch outcomes. + pub validator_up: bool, + /// The classified state of each checked prefix (already reduced to its + /// host more-specific — see [`host_more_specific`]). Only prefixes whose + /// fetch+classify succeeded are present; a prefix whose check failed is + /// omitted (its state is unknown, not assumed valid or invalid). + pub per_prefix: Vec<(ipnet::IpNet, RpkiState)>, +} + +/// Aggregate the raw fetch+classify outcome of each checked prefix into an +/// [`RpkiReport`]. +/// +/// This is deliberately the *pure* half of [`fetch::check_once`] — the I/O +/// (the actual HTTP fetch) lives in the coverage-excluded `fetch` module; +/// this function only reasons about the `Result`s that I/O already produced, +/// so it is unit-tested directly. +/// +/// **Validator-reachability rule** (documented here, not just in code, since +/// it's a judgment call): `validator_up` is `true` if AT LEAST ONE checked +/// prefix's fetch succeeded at the transport/parse level, and `false` only +/// when EVERY fetch failed — i.e. the validator is reported down only when +/// it is genuinely unreachable for the whole pass, not merely for one flaky +/// lookup among several successes. An empty prefix set has nothing to +/// disprove reachability, so it reports `validator_up: true` (nothing to +/// fail open on). A per-prefix fetch error always means that prefix's state +/// is unknown, so it's dropped from `per_prefix` rather than guessed at — +/// this per-prefix fail-open is unconditional and independent of the +/// aggregate `validator_up` verdict. +pub fn aggregate_report(results: Vec<(ipnet::IpNet, Result)>) -> RpkiReport { + let validator_up = results.is_empty() || results.iter().any(|(_, r)| r.is_ok()); + let per_prefix = results + .into_iter() + .filter_map(|(net, r)| r.ok().map(|state| (net, state))) + .collect(); + RpkiReport { + validator_up, + per_prefix, + } +} + +/// Tracks previously-observed RPKI validator reachability and per-prefix +/// validity state so the periodic checker (in `blackwalld`) can WARN only on +/// a state **transition**, never on a steady-state repeat — a standing RPKI +/// gap (e.g. a ROA with a short `maxLength`) must not spam the log every +/// `rpki-check-interval` forever. +/// +/// This is the pure, unit-tested dedup core; the periodic tokio task that +/// drives it lives in `blackwalld` (coverage-excluded I/O glue). +#[derive(Debug, Clone)] +pub struct RpkiWarnState { + validator_up: bool, + prefixes: HashMap, +} + +impl Default for RpkiWarnState { + /// Starts assuming the validator is reachable (`true`) and with no prior + /// per-prefix observations — so the very first down/invalid observation + /// after startup is treated as a transition and warns, rather than being + /// silently absorbed as "no change from an unknown baseline". + fn default() -> Self { + Self { + validator_up: true, + prefixes: HashMap::new(), + } + } +} + +impl RpkiWarnState { + /// Record the validator's reachability observed on this check pass. + /// + /// Returns `true` only on an up→down transition — "you should WARN + /// now". A down→down repeat and an up→up repeat both return `false` (no + /// change worth logging again); a down→up recovery also returns `false`, + /// but the transition is still recorded — the caller should log that + /// case as an INFO, not a WARN. + pub fn observe_validator_up(&mut self, up: bool) -> bool { + let should_warn = self.validator_up && !up; + self.validator_up = up; + should_warn + } + + /// Record the validity state observed for `net` on this check pass. + /// + /// Returns `true` — "you should WARN now" — when `state` is + /// [`RpkiState::Invalid`] or [`RpkiState::NotFound`] (a state that would + /// cause a validating upstream to drop the blackhole announcement) **and** + /// it is either the first observation of `net` or different from the + /// previously observed state for `net`. A repeat of the same bad state + /// returns `false` (already warned, don't spam). A transition to + /// [`RpkiState::Valid`] — a recovery — always returns `false`: it is + /// still recorded (so a later regression is detected as a fresh + /// transition), but recovering is good news, not a WARN; the caller + /// should log it as an INFO instead. + pub fn observe_prefix(&mut self, net: ipnet::IpNet, state: RpkiState) -> bool { + let prev = self.prefixes.insert(net, state); + state != RpkiState::Valid && prev != Some(state) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pfx(s: &str) -> ipnet::IpNet { + s.parse().expect("valid prefix") + } + + #[test] + fn host_more_specific_v4_and_v6() { + assert_eq!( + host_more_specific(&pfx("94.156.238.0/24")), + pfx("94.156.238.0/32") + ); + assert_eq!( + host_more_specific(&pfx("2a12:9b00:b00b::/48")), + pfx("2a12:9b00:b00b::/128") + ); + } + + #[test] + fn host_more_specific_is_idempotent_on_already_host_length() { + assert_eq!( + host_more_specific(&pfx("203.0.113.5/32")), + pfx("203.0.113.5/32") + ); + assert_eq!( + host_more_specific(&pfx("2001:db8::1/128")), + pfx("2001:db8::1/128") + ); + } + + #[test] + fn host_more_specific_masks_host_bits() { + // A non-network address (host bits set) must be masked down to the + // network address, not just have its prefix length changed. + assert_eq!( + host_more_specific(&pfx("94.156.238.5/24")), + pfx("94.156.238.0/32") + ); + } + + #[test] + fn classify_routinator_states() { + let valid = r#"{"validated_route":{"route":{"origin_asn":"AS214806","prefix":"94.156.238.0/32"},"validity":{"state":"valid"}}}"#; + let invalid = r#"{"validated_route":{"validity":{"state":"invalid"}}}"#; + let nf = r#"{"validated_route":{"validity":{"state":"not-found"}}}"#; + assert_eq!(classify(valid).unwrap(), RpkiState::Valid); + assert_eq!(classify(invalid).unwrap(), RpkiState::Invalid); + assert_eq!(classify(nf).unwrap(), RpkiState::NotFound); + assert!( + classify(r#"{"validated_route":{"validity":{"state":"unknown"}}}"#).is_err(), + "RIPE 'unknown' / drift → err → fail-open" + ); + assert!(classify("{}").is_err()); + } + + #[test] + fn classify_rejects_malformed_json() { + assert!(classify("not json").is_err()); + assert!(classify("").is_err()); + } + + #[test] + fn classify_rejects_missing_validity() { + assert!(classify(r#"{"validated_route":{}}"#).is_err()); + } + + #[test] + fn validity_url_form() { + assert_eq!( + validity_url("http://h:8323", 214806, &pfx("94.156.238.0/32")), + "http://h:8323/api/v1/validity/AS214806/94.156.238.0/32" + ); + } + + #[test] + fn validity_url_v6_form() { + assert_eq!( + validity_url("http://h:8323", 214806, &pfx("2a12:9b00:b00b::/128")), + "http://h:8323/api/v1/validity/AS214806/2a12:9b00:b00b::/128" + ); + } + + #[test] + fn warns_only_on_state_transition() { + let mut st = RpkiWarnState::default(); + // first observation of an invalid prefix → warn + assert!(st.observe_prefix(pfx("94.156.238.0/24"), RpkiState::Invalid)); // returns true = "should warn" + // same state next interval → no warn + assert!(!st.observe_prefix(pfx("94.156.238.0/24"), RpkiState::Invalid)); + // recovers to valid → (info, not warn) → observe returns false-for-warn but records the change + assert!(!st.observe_prefix(pfx("94.156.238.0/24"), RpkiState::Valid)); + // validator down→up→down transitions warn once each + assert!(st.observe_validator_up(false)); // up(default)→down = warn + assert!(!st.observe_validator_up(false)); // still down = no warn + assert!(!st.observe_validator_up(true)); // recovery = info, not warn + assert!(st.observe_validator_up(false)); // down again = warn + } + + #[test] + fn observe_prefix_warns_on_worse_or_different_bad_state_not_on_repeat() { + let mut st = RpkiWarnState::default(); + let net = pfx("94.156.238.0/32"); + // first bad observation warns + assert!(st.observe_prefix(net, RpkiState::NotFound)); + // same bad state repeated → no warn + assert!(!st.observe_prefix(net, RpkiState::NotFound)); + // a *different* bad state → warn (still bad, but changed) + assert!(st.observe_prefix(net, RpkiState::Invalid)); + // first observation of a prefix that is immediately valid → no warn + let other = pfx("203.0.113.0/32"); + assert!(!st.observe_prefix(other, RpkiState::Valid)); + // repeated valid → still no warn + assert!(!st.observe_prefix(other, RpkiState::Valid)); + } + + #[test] + fn observe_prefix_tracks_each_prefix_independently() { + let mut st = RpkiWarnState::default(); + let a = pfx("94.156.238.0/32"); + let b = pfx("203.0.113.5/32"); + assert!(st.observe_prefix(a, RpkiState::Invalid)); + // a different, previously-unseen prefix warns on its own first observation + assert!(st.observe_prefix(b, RpkiState::NotFound)); + // repeating `a`'s state doesn't warn, `b` is untouched by it + assert!(!st.observe_prefix(a, RpkiState::Invalid)); + } + + #[test] + fn aggregate_report_validator_up_when_first_fetch_succeeds() { + let net = pfx("94.156.238.0/32"); + let results: Vec<(ipnet::IpNet, Result)> = vec![ + (net, Ok(RpkiState::Invalid)), + (pfx("203.0.113.0/32"), Err(())), + ]; + let report = aggregate_report(results); + assert!(report.validator_up); + // the failed second prefix is omitted, not guessed at + assert_eq!(report.per_prefix, vec![(net, RpkiState::Invalid)]); + } + + #[test] + fn aggregate_report_validator_up_when_only_a_later_fetch_succeeds() { + // A single flaky lookup FIRST in the list must not report the + // validator down when a later fetch in the same pass succeeds. + let results: Vec<(ipnet::IpNet, Result)> = vec![ + (pfx("94.156.238.0/32"), Err(())), + (pfx("203.0.113.0/32"), Ok(RpkiState::Valid)), + ]; + let report = aggregate_report(results); + assert!( + report.validator_up, + "at least one fetch succeeded, so the validator is up" + ); + assert_eq!( + report.per_prefix, + vec![(pfx("203.0.113.0/32"), RpkiState::Valid)] + ); + } + + #[test] + fn aggregate_report_validator_down_only_when_every_fetch_fails() { + let results: Vec<(ipnet::IpNet, Result)> = vec![ + (pfx("94.156.238.0/32"), Err(())), + (pfx("203.0.113.0/32"), Err(())), + ]; + let report = aggregate_report(results); + assert!( + !report.validator_up, + "every fetch failed, so the validator is genuinely unreachable" + ); + assert!(report.per_prefix.is_empty()); + } + + #[test] + fn aggregate_report_empty_prefix_set_is_validator_up() { + let report: RpkiReport = + aggregate_report(Vec::<(ipnet::IpNet, Result)>::new()); + assert!(report.validator_up); + assert!(report.per_prefix.is_empty()); + } +} diff --git a/crates/blackwall-rtbh/Cargo.toml b/crates/blackwall-rtbh/Cargo.toml index df7e556..368e650 100644 --- a/crates/blackwall-rtbh/Cargo.toml +++ b/crates/blackwall-rtbh/Cargo.toml @@ -19,6 +19,10 @@ thiserror = { workspace = true } # installs its own SIGUSR1 handler, mirroring blackwalld's disarm_signal_task) # — not by any unit test, so it stays scoped to dev-dependencies. tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time", "signal"] } +# Test-only counting `Layer` for the `disarm` log-once assertion (#193 +# residual) — avoids adding `tracing-test` as a new dependency; reuses the +# workspace-pinned version already used by `blackwalld`. +tracing-subscriber = { workspace = true } [lints] workspace = true diff --git a/crates/blackwall-rtbh/src/flowspec_controller.rs b/crates/blackwall-rtbh/src/flowspec_controller.rs index 853d17d..82bdb40 100644 --- a/crates/blackwall-rtbh/src/flowspec_controller.rs +++ b/crates/blackwall-rtbh/src/flowspec_controller.rs @@ -290,6 +290,21 @@ impl FlowSpecController { .collect() } + /// Look up the CURRENTLY active rule for `key`, rather than trusting a + /// possibly-stale snapshot captured elsewhere and earlier (e.g. by a + /// queued [`crate::flowspec_manager::FlowSpecManager`] reapply retry). + /// + /// `key`'s identity (destination/protocol/port) excludes the action + /// (C4), so a re-assert can legitimately change the action in place + /// (e.g. `manual_add`'s `changed_action_re_announces`) while the key + /// stays the same — a caller that wants "the rule as it stands right + /// now" must call this rather than replay a captured `FlowSpecRule`. + /// `None` if `key` is no longer active. + #[must_use] + pub fn active_rule(&self, key: FlowKey) -> Option { + self.active.get(&key).map(|e| e.rule.clone()) + } + /// Re-assert every active rule for `target` without re-announcing. /// /// Used for [`blackwall_flow::FlowMitigationEvent::Update`], which reports that diff --git a/crates/blackwall-rtbh/src/flowspec_manager.rs b/crates/blackwall-rtbh/src/flowspec_manager.rs index 6988e7a..ac5d360 100644 --- a/crates/blackwall-rtbh/src/flowspec_manager.rs +++ b/crates/blackwall-rtbh/src/flowspec_manager.rs @@ -69,6 +69,38 @@ impl MirrorOp { } } +/// A [`FlowSpecManager::rehydrate`] re-announce that failed at the +/// [`BgpExecutor`] and is queued for a self-heal retry (issue #194). +/// +/// Mirrors [`crate::manager`]'s private `ReapplyOp`, keyed by [`FlowKey`] +/// instead of target IP — unlike [`MirrorOp`] (which only ever replays a +/// journal write, the BGP side already having succeeded), a queued +/// `ReapplyOp` re-attempts the BGP `announce_flowspec` itself: rehydrate's +/// failure happens on the BGP side, not the journal side (rehydrate never +/// journals in the first place). +/// +/// Holds only the [`FlowKey`], not the rule that was captured when the op +/// was queued: [`FlowSpecManager::retry_pending_reapply`] re-derives the +/// CURRENT rule from [`FlowSpecController::active_rule`] at retry time +/// rather than replaying a snapshot, so a fresh, successful re-assertion of +/// the same key with a changed action (C4 — e.g. `manual_add`'s +/// `changed_action_re_announces`) that lands between the failed rehydrate +/// and the retry is never clobbered by the stale queued content (#194 C1 +/// follow-up). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ReapplyOp { + /// The key of the rule to re-announce; the current rule content is + /// looked up fresh from the controller at retry time. + key: FlowKey, +} + +impl ReapplyOp { + /// The `FlowKey` this reapply op concerns. + fn key(&self) -> FlowKey { + self.key + } +} + /// Outcome of [`FlowSpecManager::execute_and_journal_announce`]. /// /// Mirrors [`crate::manager`]'s private `AnnounceOutcome`. The auto path @@ -119,6 +151,16 @@ pub struct FlowSpecManager { /// succeeded; retried (never re-issued to BGP) by /// `FlowSpecManager::retry_pending_mirror` on the next tick. pending_mirror: Vec, + /// `rehydrate` re-announces that failed at the [`BgpExecutor`]; retried + /// by [`Self::retry_pending_reapply`] on the next tick (issue #194). The + /// controller's active entry from [`FlowSpecController::resume`] is kept + /// (never rolled back) while queued — unlike a live BGP failure on the + /// `apply_open`/`apply_add` path, a rehydrated rule is a known-good + /// persisted mitigation with no fresh detection to naturally re-attempt + /// it, so rollback would strand the control plane believing nothing is + /// announced while the journal still says otherwise (mirrors + /// `blackwall_rtbh::manager::RtbhManager`'s private `pending_reapply`). + pending_reapply: Vec, /// Count of announces that failed at the BGP executor, each rolled back /// (see [`Self::apply_failures`]). apply_failures: u64, @@ -155,6 +197,7 @@ impl FlowSpecManager { bgp, journal, pending_mirror: Vec::new(), + pending_reapply: Vec::new(), apply_failures: 0, rate_limiter: None, ratecapped: 0, @@ -233,10 +276,14 @@ impl FlowSpecManager { /// /// 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. + /// `FlowSpecManager::retry_pending_mirror`), then any `rehydrate` + /// re-announces queued by a previous tick's transient BGP failure (see + /// [`Self::retry_pending_reapply`], issue #194), so both self-heals + /// converge within one tick interval of the respective dependency + /// recovering. pub async fn tick(&mut self, mono_now: u64, wall_now: u64) { self.retry_pending_mirror().await; + self.retry_pending_reapply().await; let actions = self.controller.tick(mono_now); for action in actions { self.execute_and_journal(action, mono_now, wall_now).await; @@ -328,6 +375,12 @@ impl FlowSpecManager { /// /// For each row, calls [`FlowSpecController::resume`] and re-announces on /// BGP (without journaling — the row already exists in the journal). If + /// the re-announce fails, the controller's entry is kept active (it is a + /// known-good persisted mitigation, not rolled back the way a fresh + /// `apply_open`/`apply_add` failure is — see the module docs) and queued + /// via [`Self::queue_reapply`] for a retry on the next [`Self::tick`] + /// (issue #194): unlike a live detection, a rehydrated rule has no + /// natural re-detection to compensate for a dropped announce. 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( @@ -339,8 +392,9 @@ impl FlowSpecManager { 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"); + if let Err(e) = self.bgp.announce_flowspec(r.clone()).await { + tracing::warn!(%target, error = %e, "FlowSpec: rehydrate re-announce failed; queuing for retry"); + self.queue_reapply(ReapplyOp { key: key_of(&r) }); } continue; } @@ -392,6 +446,18 @@ impl FlowSpecManager { self.ratecapped } + /// Number of `rehydrate` re-announces currently queued for a self-heal + /// retry after a failed BGP announce on restart (issue #194). Each + /// queued rule is still active in the controller (kept, not rolled back) + /// but not yet confirmed on the wire; drained by + /// [`Self::retry_pending_reapply`] on the next [`Self::tick`]. Surfaced + /// for `/metrics` as `blackwall_flowspec_reapply_pending`, mirroring + /// `blackwall_rtbh::manager::RtbhManager::reapply_pending`. + #[must_use] + pub fn reapply_pending(&self) -> usize { + self.pending_reapply.len() + } + /// In-daemon disarm kill switch (C5): withdraw every currently-active /// rule and switch to record-only for the rest of this process's life. /// @@ -464,6 +530,18 @@ impl FlowSpecManager { self.pending_mirror.push(op); } + /// Queue a failed `rehydrate` re-announce for retry, coalescing by + /// [`FlowKey`] (issue #194). + /// + /// Mirrors [`Self::queue_mirror`]'s coalescing: only the latest queued + /// op per key is kept, since a repeat rehydrate failure for the same + /// rule during an outage should never grow the queue past one entry. + fn queue_reapply(&mut self, op: ReapplyOp) { + let key = op.key(); + self.pending_reapply.retain(|o| o.key() != key); + self.pending_reapply.push(op); + } + /// Execute one controller action on BGP and mirror it into the journal. async fn execute_and_journal(&mut self, action: FlowSpecAction, mono_now: u64, wall_now: u64) { match action { @@ -583,6 +661,44 @@ impl FlowSpecManager { } } + /// Drain-retry queued `rehydrate` re-announces left over from a + /// transient BGP failure (issue #194). + /// + /// Each queued op re-derives the CURRENT rule for its key from + /// [`FlowSpecController::active_rule`] rather than replaying the rule + /// snapshot captured when the op was queued: if the key is no longer + /// active (e.g. cleared by a manual remove or a hold-down expiry between + /// the failed rehydrate and this tick), `active_rule` returns `None` and + /// the op is dropped — re-announcing a rule the control plane no longer + /// wants live would itself create a phantom. If the key is still active + /// but a fresh, successful re-assertion changed its action in the + /// meantime (C4 — e.g. a detection or operator call tightening the + /// rate), re-deriving picks up that CURRENT action instead of replaying + /// the stale queued one, which would otherwise silently revert the fresh + /// update (#194 C1 follow-up). Otherwise the announce is re-attempted; + /// ops that still fail are kept (retried again on the next call), ops + /// that succeed are dropped. + async fn retry_pending_reapply(&mut self) { + if self.pending_reapply.is_empty() { + return; + } + let ops = std::mem::take(&mut self.pending_reapply); + for op in ops { + let key = op.key(); + let Some(rule) = self.controller.active_rule(key) else { + tracing::info!( + ?key, + "FlowSpec: dropping queued rehydrate reapply; entry no longer active" + ); + continue; + }; + if let Err(e) = self.bgp.announce_flowspec(rule).await { + tracing::warn!(?key, error = %e, "FlowSpec: rehydrate reapply retry failed; re-queuing"); + self.pending_reapply.push(op); + } + } + } + #[cfg(test)] pub(crate) fn bgp(&self) -> &B { &self.bgp @@ -609,11 +725,43 @@ mod tests { use std::sync::Mutex; use std::time::Duration; - #[derive(Default)] + #[derive(Default, Clone)] struct FakeBgp { - announced: Mutex>, - withdrawn: Mutex>, - fail: bool, + announced: Arc>>, + withdrawn: Arc>>, + fail: Arc>, + /// Independent withdraw-only failure toggle, for exercising disarm's + /// best-effort tolerance of a withdraw `Err` without also blocking + /// the announce that must precede it (unlike `fail`, which fails + /// both). Mirrors `crate::manager::RtbhManager`'s test fake. + fail_withdraw: Arc>, + } + impl FakeBgp { + /// Build a fake whose `announce_flowspec`/`withdraw_flowspec` fail + /// from the start. + fn with_fail(fail: bool) -> Self { + let f = Self::default(); + f.set_fail(fail); + f + } + /// Build a fake whose `withdraw_flowspec` alone fails from the start + /// (`announce_flowspec` still succeeds). + fn with_fail_withdraw(fail_withdraw: bool) -> Self { + let f = Self::default(); + *f.fail_withdraw.lock().unwrap() = fail_withdraw; + f + } + /// Flip the announce/withdraw failure toggle at runtime — lets a + /// test simulate a BGP session recovering mid-scenario (a clone + /// shares the same underlying flag with whatever manager holds this + /// fake). + fn set_fail(&self, fail: bool) { + *self.fail.lock().unwrap() = fail; + } + /// Whether `rule` was ever announced. + fn announced_contains(&self, rule: &FlowSpecRule) -> bool { + self.announced.lock().unwrap().contains(rule) + } } #[async_trait] impl BgpExecutor for FakeBgp { @@ -630,7 +778,7 @@ mod tests { &self, rule: FlowSpecRule, ) -> Result<(), crate::manager::BgpError> { - if self.fail { + if *self.fail.lock().unwrap() { return Err(crate::manager::BgpError); } self.announced.lock().unwrap().push(rule); @@ -640,7 +788,7 @@ mod tests { &self, rule: FlowSpecRule, ) -> Result<(), crate::manager::BgpError> { - if self.fail { + if *self.fail.lock().unwrap() || *self.fail_withdraw.lock().unwrap() { return Err(crate::manager::BgpError); } self.withdrawn.lock().unwrap().push(rule); @@ -724,10 +872,7 @@ mod tests { fn mgr(fail_bgp: bool, fail_j: bool) -> FlowSpecManager { FlowSpecManager::new( FlowSpecController::new(cfg()), - FakeBgp { - fail: fail_bgp, - ..Default::default() - }, + FakeBgp::with_fail(fail_bgp), FakeJournal { fail: fail_j, ..Default::default() @@ -1214,6 +1359,61 @@ mod tests { assert_eq!(m.disarmed_skips(), 1); } + #[tokio::test] + async fn disarm_tolerates_a_withdraw_error() { + // Best-effort: a withdraw_flowspec Err during disarm must not abort + // the sweep (a second active rule is still withdrawn) or stop the + // manager from switching to record-only. Mirrors + // `crate::manager::RtbhManager`'s `disarm_tolerates_a_withdraw_error`. + let mut m = FlowSpecManager::new( + FlowSpecController::new(cfg()), + FakeBgp::with_fail_withdraw(true), + FakeJournal::default(), + ); + m.apply_open( + ip("203.0.113.7"), + &[flow_rule("203.0.113.7", 17, 53, 0.0)], + 0, + 0, + ) + .await; + m.apply_open( + ip("203.0.113.8"), + &[flow_rule("203.0.113.8", 17, 53, 0.0)], + 0, + 0, + ) + .await; + let key1 = key_of(&rule("203.0.113.7/32", 17, 53, 0.0)); + let key2 = key_of(&rule("203.0.113.8/32", 17, 53, 0.0)); + assert!(m.is_active(key1)); + assert!(m.is_active(key2)); + + m.disarm(1_000).await; + + assert!( + m.bgp().withdrawn.lock().unwrap().is_empty(), + "every withdraw errored, so none was recorded by the fake" + ); + assert!(!m.is_active(key1)); + assert!( + !m.is_active(key2), + "disarm clears the active set even when every withdraw errors (best-effort)" + ); + + // Record-only holds even though disarm itself never got a + // confirmed withdraw. + m.apply_open( + ip("203.0.113.9"), + &[flow_rule("203.0.113.9", 17, 53, 0.0)], + 2_000, + 2_000, + ) + .await; + let key3 = key_of(&rule("203.0.113.9/32", 17, 53, 0.0)); + assert!(!m.is_active(key3)); + } + #[tokio::test] async fn apply_add_while_disarmed_is_rejected_not_applied() { // C5 + final-review fix: a manual add while disarmed must be @@ -1305,4 +1505,111 @@ mod tests { assert!(m.journal().announced.lock().unwrap().is_empty()); assert_eq!(m.journal().withdrawn.lock().unwrap().len(), 1); } + + #[tokio::test] + async fn rehydrate_failure_queues_reapply_and_tick_reconverges() { + // #194 C1: a persisted, eligible rule whose rehydrate re-announce + // fails must NOT be left stranded (active in the controller but + // never on the wire) — it is queued for retry and the queue drains + // once the BGP session recovers. + let bgp = FakeBgp::with_fail(true); // announce errors + let mut m = FlowSpecManager::new( + FlowSpecController::new(cfg()), + bgp.clone(), + FakeJournal::default(), + ); + let r = rule("203.0.113.7/32", 17, 53, 0.0); + m.rehydrate(vec![(r.clone(), 1_000, BlackholeOrigin::Auto)], 1_000) + .await; + let key = key_of(&r); + assert!(m.is_active(key), "entry kept (not dropped)"); + assert_eq!( + m.reapply_pending(), + 1, + "failed re-announce queued for retry" + ); + + // Session recovers; the next tick re-announces and drains the queue. + bgp.set_fail(false); + m.tick(2_000, 2_000).await; + assert_eq!(m.reapply_pending(), 0); + assert!(bgp.announced_contains(&r)); + } + + #[tokio::test] + async fn queue_reapply_dedupes_and_drops_if_cleared() { + // A still-failing tick must coalesce (not double-enqueue) the retry + // for the same rule; and once the rule is cleared before it ever + // succeeds, the queued reapply must be dropped rather than + // re-announcing a no-longer-wanted rule. + let bgp = FakeBgp::with_fail(true); + let mut m = FlowSpecManager::new( + FlowSpecController::new(cfg()), + bgp.clone(), + FakeJournal::default(), + ); + let r = rule("203.0.113.7/32", 17, 53, 0.0); + m.rehydrate(vec![(r.clone(), 1_000, BlackholeOrigin::Auto)], 1_000) + .await; + m.tick(2_000, 2_000).await; // still failing -> re-queued, NOT double-enqueued + assert_eq!(m.reapply_pending(), 1, "coalesced, not doubled"); + + // Cleared before it ever succeeded (manual withdraw). + m.apply_remove(r.clone(), 3_000, 3_000).await; + bgp.set_fail(false); + m.tick(4_000, 4_000).await; + assert_eq!(m.reapply_pending(), 0, "dropped: entry no longer active"); + assert!(!bgp.announced_contains(&r), "not re-announced after clear"); + } + + #[tokio::test] + async fn stale_reapply_does_not_clobber_a_fresh_successful_update() { + // #194 C1 follow-up: `retry_pending_reapply` must re-derive the + // controller's CURRENT rule for the key at retry time, not replay + // the rule snapshot captured when the op was queued. Otherwise a + // fresh, successful re-assertion of the SAME key with a CHANGED + // action (C4 — a re-assert may legitimately change the action, see + // `manual_add_upgrade_with_changed_action_re_announces`) that lands + // between the failed rehydrate and the retry gets silently reverted + // by the stale queued content — a mitigation-weakening regression. + let bgp = FakeBgp::with_fail(true); // rehydrate re-announce fails + let mut m = FlowSpecManager::new( + FlowSpecController::new(cfg()), + bgp.clone(), + FakeJournal::default(), + ); + let old = rule("203.0.113.7/32", 17, 53, 500.0); + m.rehydrate(vec![(old.clone(), 1_000, BlackholeOrigin::Auto)], 1_000) + .await; + let key = key_of(&old); + assert!(m.is_active(key), "entry kept (not dropped)"); + assert_eq!(m.reapply_pending(), 1, "failed rehydrate queued for retry"); + + // BGP recovers just in time for a FRESH, successful re-assertion + // with a tighter (different) action for the SAME key, landing + // before the next tick drains the stale queue. + bgp.set_fail(false); + let new = rule("203.0.113.7/32", 17, 53, 100.0); + let outcome = m.apply_add(new.clone(), 1_500, 1_500).await; + assert_eq!(outcome, ApplyOutcome::Applied); + assert!( + bgp.announced_contains(&new), + "the fresh, tighter update reached BGP" + ); + + // The next tick drains the (now-stale) queued reapply. It must + // re-derive and re-announce the CURRENT rule (rate 100.0), not + // replay the stale queued snapshot (rate 500.0) — which would + // silently revert the fresh tightening. + m.tick(2_000, 2_000).await; + assert_eq!(m.reapply_pending(), 0); + + let announced = m.bgp().announced.lock().unwrap(); + let last = announced.last().expect("at least one announce recorded"); + assert_eq!( + last.action, + FlowAction::TrafficRate(100.0), + "the drained retry must re-announce the CURRENT rule, not the stale queued one" + ); + } } diff --git a/crates/blackwall-rtbh/src/manager.rs b/crates/blackwall-rtbh/src/manager.rs index f35deb5..ee85d99 100644 --- a/crates/blackwall-rtbh/src/manager.rs +++ b/crates/blackwall-rtbh/src/manager.rs @@ -99,6 +99,15 @@ pub struct RtbhManager { /// succeeded; retried (never re-issued to BGP) by /// `RtbhManager::retry_pending_mirror` on the next tick. pending_mirror: Vec, + /// `rehydrate` re-announces that failed at the [`BgpExecutor`]; retried + /// by [`Self::retry_pending_reapply`] on the next tick (issue #194). The + /// controller's active entry from [`RtbhController::resume`] is kept + /// (never rolled back) while queued — unlike a live BGP failure on the + /// `apply_event`/`apply_add` path (see the module docs), a rehydrated row + /// is a known-good persisted mitigation with no fresh detection to + /// naturally re-attempt it, so rollback would strand the control plane + /// believing nothing is announced while the journal still says otherwise. + pending_reapply: Vec, /// Count of announces that failed at the BGP executor, each rolled back /// (see [`Self::apply_failures`]). apply_failures: u64, @@ -153,6 +162,21 @@ impl MirrorOp { } } +/// A [`RtbhManager::rehydrate`] re-announce that failed at the +/// [`BgpExecutor`] and is queued for a self-heal retry (issue #194). +/// +/// Unlike [`MirrorOp`] (which only ever replays a journal write, the BGP +/// side already having succeeded), a queued `ReapplyOp` re-attempts the BGP +/// `announce` itself — rehydrate's failure happens on the BGP side, not the +/// journal side (rehydrate never journals in the first place). +#[derive(Debug, Clone, PartialEq, Eq)] +struct ReapplyOp { + /// The blackhole target this re-announce concerns. + target: IpAddr, + /// The route to re-announce. + route: Route, +} + /// Outcome of [`RtbhManager::execute_and_journal_announce`]. /// /// The auto path (`apply_event`/`tick`, via [`RtbhManager::execute_and_journal`]) @@ -185,6 +209,7 @@ impl RtbhManager { bgp, journal, pending_mirror: Vec::new(), + pending_reapply: Vec::new(), apply_failures: 0, rate_limiter: None, ratecapped: 0, @@ -227,10 +252,13 @@ impl RtbhManager { /// /// Starts by retrying any journal mirror writes queued by a previous /// tick's transient failure (see `RtbhManager::retry_pending_mirror`), - /// so a self-heal converges within one tick interval of the DB - /// recovering. + /// then any `rehydrate` re-announces queued by a previous tick's + /// transient BGP failure (see `RtbhManager::retry_pending_reapply`, + /// issue #194), so both self-heals converge within one tick interval of + /// the respective dependency recovering. pub async fn tick(&mut self, mono_now: u64, wall_now: u64) { self.retry_pending_mirror().await; + self.retry_pending_reapply().await; let actions = self.controller.tick(mono_now); for action in actions { self.execute_and_journal(action, mono_now, wall_now).await; @@ -328,7 +356,13 @@ impl RtbhManager { /// Re-install persisted blackholes on a fresh session (rehydration). /// /// For each row, calls [`RtbhController::resume`] and re-announces on BGP - /// (without journaling — the row already exists in the journal). If + /// (without journaling — the row already exists in the journal). If the + /// re-announce fails, the controller's entry is kept active (it is a + /// known-good persisted mitigation, not rolled back the way a fresh + /// `apply_event`/`apply_add` failure is — see the module docs) and + /// queued via [`Self::queue_reapply`] for a retry on the next + /// [`Self::tick`] (issue #194): unlike a live detection, a rehydrated row + /// has no natural re-detection to compensate for a dropped announce. If /// `resume` returns no action (over cap, ineligible, or no next-hop), /// this logs a warning naming the target; a row is never silently /// dropped. @@ -336,8 +370,9 @@ impl RtbhManager { for (target, _persisted_at, origin) in rows { let actions = self.controller.resume(target, mono_now, origin); if let Some(RtbhAction::Announce(route)) = actions.into_iter().next() { - if let Err(e) = self.bgp.announce(route).await { - tracing::warn!(%target, error = %e, "RTBH: rehydrate re-announce failed"); + if let Err(e) = self.bgp.announce(route.clone()).await { + tracing::warn!(%target, error = %e, "RTBH: rehydrate re-announce failed; queuing for retry"); + self.queue_reapply(ReapplyOp { target, route }); } continue; } @@ -392,6 +427,18 @@ impl RtbhManager { self.ratecapped } + /// Number of `rehydrate` re-announces currently queued for a self-heal + /// retry after a failed BGP announce on restart (issue #194). Each + /// queued target is still active in the controller (kept, not rolled + /// back) but not yet confirmed on the wire; drained by + /// [`Self::retry_pending_reapply`] on the next [`Self::tick`]. Surfaced + /// for `/metrics` as `blackwall_rtbh_reapply_pending`, mirroring how + /// [`Self::apply_failures`] reaches the endpoint. + #[must_use] + pub fn reapply_pending(&self) -> usize { + self.pending_reapply.len() + } + /// In-daemon disarm kill switch (C5): withdraw every currently-active /// blackhole and switch to record-only for the rest of this process's /// life. @@ -465,6 +512,17 @@ impl RtbhManager { self.pending_mirror.push(op); } + /// Queue a failed `rehydrate` re-announce for retry, coalescing by + /// target (issue #194). + /// + /// Mirrors [`Self::queue_mirror`]'s coalescing: only the latest queued + /// op per target is kept, since a repeat rehydrate failure for the same + /// target during an outage should never grow the queue past one entry. + fn queue_reapply(&mut self, op: ReapplyOp) { + self.pending_reapply.retain(|o| o.target != op.target); + self.pending_reapply.push(op); + } + /// Execute one controller action on BGP and mirror it into the journal. async fn execute_and_journal(&mut self, action: RtbhAction, mono_now: u64, wall_now: u64) { match action { @@ -581,6 +639,33 @@ impl RtbhManager { } } + /// Drain-retry queued `rehydrate` re-announces left over from a + /// transient BGP failure (issue #194). + /// + /// Each queued op is first re-checked against the current active set: + /// if the target is no longer active (e.g. cleared by a manual remove + /// or a hold-down expiry between the failed rehydrate and this tick), + /// the op is dropped — re-announcing a route the control plane no + /// longer wants live would itself create a phantom. Otherwise the + /// announce is re-attempted; ops that still fail are kept (retried + /// again on the next call), ops that succeed are dropped. + async fn retry_pending_reapply(&mut self) { + if self.pending_reapply.is_empty() { + return; + } + let ops = std::mem::take(&mut self.pending_reapply); + for op in ops { + if !self.is_active(op.target) { + tracing::info!(target = %op.target, "RTBH: dropping queued rehydrate reapply; entry no longer active"); + continue; + } + if let Err(e) = self.bgp.announce(op.route.clone()).await { + tracing::warn!(target = %op.target, error = %e, "RTBH: rehydrate reapply retry failed; re-queuing"); + self.pending_reapply.push(op); + } + } + } + #[cfg(test)] pub(crate) fn bgp(&self) -> &B { &self.bgp @@ -606,34 +691,60 @@ fn ip_of(prefix: &IpNet) -> IpAddr { #[cfg(test)] mod tests { use super::*; - use crate::{BlackholeOrigin, RtbhConfig, RtbhController}; + use crate::{BlackholeOrigin, NoOpJournal, RtbhConfig, RtbhController}; use blackwall_flow::{AttackKind, Detection, DetectionEvent, Severity}; use std::net::IpAddr; use std::sync::Mutex; use std::time::Duration; - #[derive(Default)] + #[derive(Default, Clone)] struct FakeBgp { - announced: Mutex>, - withdrawn: Mutex>, - fail: bool, + announced: Arc>>, + withdrawn: Arc>>, + fail: Arc>, /// Independent withdraw-only failure toggle, for exercising disarm's /// best-effort tolerance of a withdraw `Err` without also blocking /// the announce that must precede it (unlike `fail`, which fails /// both). - fail_withdraw: bool, + fail_withdraw: Arc>, + } + impl FakeBgp { + /// Build a fake whose `announce`/`withdraw` fail from the start. + fn with_fail(fail: bool) -> Self { + let f = Self::default(); + f.set_fail(fail); + f + } + /// Build a fake whose `withdraw` alone fails from the start + /// (`announce` still succeeds). + fn with_fail_withdraw(fail_withdraw: bool) -> Self { + let f = Self::default(); + *f.fail_withdraw.lock().unwrap() = fail_withdraw; + f + } + /// Flip the announce/withdraw failure toggle at runtime — lets a test + /// simulate a BGP session recovering mid-scenario (a clone shares the + /// same underlying flag with whatever manager holds this fake). + fn set_fail(&self, fail: bool) { + *self.fail.lock().unwrap() = fail; + } + /// Whether `prefix` (e.g. `"203.0.113.7/32"`) was ever announced. + fn announced_contains(&self, prefix: &str) -> bool { + let net: IpNet = prefix.parse().expect("valid prefix in test"); + self.announced.lock().unwrap().contains(&net) + } } #[async_trait] impl BgpExecutor for FakeBgp { async fn announce(&self, route: Route) -> Result<(), BgpError> { - if self.fail { + if *self.fail.lock().unwrap() { return Err(BgpError); } self.announced.lock().unwrap().push(route.prefix); Ok(()) } async fn withdraw(&self, prefix: IpNet) -> Result<(), BgpError> { - if self.fail || self.fail_withdraw { + if *self.fail.lock().unwrap() || *self.fail_withdraw.lock().unwrap() { return Err(BgpError); } self.withdrawn.lock().unwrap().push(prefix); @@ -647,7 +758,7 @@ mod tests { &self, _rule: blackwall_bgp::FlowSpecRule, ) -> Result<(), BgpError> { - if self.fail { + if *self.fail.lock().unwrap() { return Err(BgpError); } Ok(()) @@ -656,7 +767,7 @@ mod tests { &self, _rule: blackwall_bgp::FlowSpecRule, ) -> Result<(), BgpError> { - if self.fail { + if *self.fail.lock().unwrap() { return Err(BgpError); } Ok(()) @@ -737,13 +848,16 @@ mod tests { fn ip(s: &str) -> IpAddr { s.parse().unwrap() } + /// A fresh controller over the same eligible-prefix config `mgr` uses — + /// named for readability at rehydrate-focused call sites that build a + /// [`RtbhManager`] directly rather than through `mgr`. + fn controller_eligible() -> RtbhController { + RtbhController::new(cfg()) + } fn mgr(fail_bgp: bool, fail_j: bool) -> RtbhManager { RtbhManager::new( RtbhController::new(cfg()), - FakeBgp { - fail: fail_bgp, - ..Default::default() - }, + FakeBgp::with_fail(fail_bgp), FakeJournal { fail: fail_j, ..Default::default() @@ -1201,10 +1315,7 @@ mod tests { // manager from switching to record-only. let mut m = RtbhManager::new( RtbhController::new(cfg()), - FakeBgp { - fail_withdraw: true, - ..Default::default() - }, + FakeBgp::with_fail_withdraw(true), FakeJournal::default(), ); m.apply_event(&DetectionEvent::Opened(det("203.0.113.7")), 0, 0) @@ -1231,6 +1342,76 @@ mod tests { assert!(!m.is_active(ip("203.0.113.9"))); } + /// Counts `tracing` events whose formatted message contains `DISARMED`. + /// + /// Test-only counting [`Layer`] standing in for `tracing-test` (not a + /// workspace dependency — see the `blackwall-rtbh/Cargo.toml` dev-dep + /// comment): guards the #193 final-review fix that `disarm()` logs its + /// banner exactly once. Before that fix, `bin/blackwalld/src/main.rs` + /// ALSO logged a pre-call `DISARMED` warn in each manager task's + /// `select!` arm, ahead of `disarm()` actually running — a second, + /// premature copy of the same banner. This test only exercises + /// `RtbhManager::disarm` directly (the manager-side log), so it can't see + /// that main.rs duplicate; it exists to pin the manager side at exactly + /// 1 so a future regression there is caught too. + #[derive(Clone, Default)] + struct DisarmedCounter(Arc); + + impl tracing_subscriber::Layer for DisarmedCounter { + fn on_event( + &self, + event: &tracing::Event<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + struct MessageContains<'a> { + needle: &'a str, + found: bool, + } + impl tracing::field::Visit for MessageContains<'_> { + fn record_debug( + &mut self, + field: &tracing::field::Field, + value: &dyn std::fmt::Debug, + ) { + if field.name() == "message" && format!("{value:?}").contains(self.needle) { + self.found = true; + } + } + } + let mut visitor = MessageContains { + needle: "DISARMED", + found: false, + }; + event.record(&mut visitor); + if visitor.found { + self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + } + } + + #[tokio::test] + async fn disarm_logs_disarmed_exactly_once() { + use tracing_subscriber::layer::SubscriberExt as _; + + let counter = DisarmedCounter::default(); + let subscriber = tracing_subscriber::registry().with(counter.clone()); + let _guard = tracing::subscriber::set_default(subscriber); + + let mut m = mgr(false, false); + m.apply_event(&DetectionEvent::Opened(det("203.0.113.7")), 0, 0) + .await; + + // A repeated disarm is a no-op (idempotent) and must not re-log. + m.disarm(1_000).await; + m.disarm(2_000).await; + + assert_eq!( + counter.0.load(std::sync::atomic::Ordering::SeqCst), + 1, + "DISARMED must be logged exactly once by the manager, even across a repeated disarm() call" + ); + } + #[tokio::test] async fn apply_add_while_disarmed_is_rejected_not_applied() { // C5 + final-review fix: a manual add while disarmed must be @@ -1311,4 +1492,60 @@ mod tests { "self-heal recorded the Manual upgrade, not the stale Auto origin" ); } + + #[tokio::test] + async fn rehydrate_failure_queues_reapply_and_tick_reconverges() { + // #194 C1: a persisted, eligible row whose rehydrate re-announce + // fails must NOT be left stranded (active in the controller but + // never on the wire) — it is queued for retry and the queue drains + // once the BGP session recovers. + let bgp = FakeBgp::default(); + bgp.set_fail(true); // announce errors + let mut mgr = RtbhManager::new(controller_eligible(), bgp.clone(), NoOpJournal); + mgr.rehydrate( + vec![(ip("203.0.113.7"), 1_000, BlackholeOrigin::Auto)], + 1_000, + ) + .await; + assert!(mgr.is_active(ip("203.0.113.7")), "entry kept (not dropped)"); + assert_eq!( + mgr.reapply_pending(), + 1, + "failed re-announce queued for retry" + ); + + // Session recovers; the next tick re-announces and drains the queue. + bgp.set_fail(false); + mgr.tick(2_000, 2_000).await; + assert_eq!(mgr.reapply_pending(), 0); + assert!(bgp.announced_contains("203.0.113.7/32")); + } + + #[tokio::test] + async fn queue_reapply_dedupes_and_drops_if_cleared() { + // A still-failing tick must coalesce (not double-enqueue) the retry + // for the same target; and once the entry is cleared before it ever + // succeeds, the queued reapply must be dropped rather than + // re-announcing a no-longer-wanted route. + let bgp = FakeBgp::default(); + bgp.set_fail(true); + let mut mgr = RtbhManager::new(controller_eligible(), bgp.clone(), NoOpJournal); + mgr.rehydrate( + vec![(ip("203.0.113.7"), 1_000, BlackholeOrigin::Auto)], + 1_000, + ) + .await; + mgr.tick(2_000, 2_000).await; // still failing -> re-queued, NOT double-enqueued + assert_eq!(mgr.reapply_pending(), 1, "coalesced, not doubled"); + + // Cleared before it ever succeeded (manual withdraw). + mgr.apply_remove(ip("203.0.113.7"), 3_000, 3_000).await; + bgp.set_fail(false); + mgr.tick(4_000, 4_000).await; + assert_eq!(mgr.reapply_pending(), 0, "dropped: entry no longer active"); + assert!( + !bgp.announced_contains("203.0.113.7/32"), + "not re-announced after clear" + ); + } } diff --git a/crates/blackwall-state/src/lib.rs b/crates/blackwall-state/src/lib.rs index 66a68f5..48de377 100644 --- a/crates/blackwall-state/src/lib.rs +++ b/crates/blackwall-state/src/lib.rs @@ -1893,6 +1893,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } @@ -1933,6 +1935,8 @@ mod tests { stateless_tcp_ports: Vec::new(), protected_prefixes: Vec::new(), shadow: false, + rpki_validator: None, + rpki_check_interval: std::time::Duration::from_secs(3600), } } diff --git a/crates/blackwall-xdp/src/control.rs b/crates/blackwall-xdp/src/control.rs index 20bc492..7851379 100644 --- a/crates/blackwall-xdp/src/control.rs +++ b/crates/blackwall-xdp/src/control.rs @@ -303,6 +303,31 @@ impl XdpController { self.rate_limited.contains_key(&src) } + /// Look up the CURRENTLY effective rate-limit action for `src`, rather + /// than trusting a possibly-stale snapshot captured elsewhere and + /// earlier (e.g. by a queued [`crate::manager::XdpManager`] reapply + /// retry). `None` if `src` is no longer rate-limited. + #[must_use] + pub fn current_rate_limit(&self, src: IpAddr) -> Option { + self.rate_limited.get(&src).map(|e| XdpAction::RateLimit { + src, + pps: e.pps, + burst: e.burst, + victim: e.victim, + }) + } + + /// Look up whether `net` is CURRENTLY blocked, rather than trusting a + /// possibly-stale snapshot captured elsewhere and earlier. `None` if + /// `net` is no longer blocked. See [`Self::current_rate_limit`] for why + /// this is exposed. + #[must_use] + pub fn current_block(&self, net: IpNet) -> Option { + self.blocked_nets + .contains_key(&net) + .then_some(XdpAction::Block { net }) + } + /// Undo a just-inserted active entry after its executor apply failed /// (C2: commit-after-confirm). /// diff --git a/crates/blackwall-xdp/src/manager.rs b/crates/blackwall-xdp/src/manager.rs index 0e584f5..3759323 100644 --- a/crates/blackwall-xdp/src/manager.rs +++ b/crates/blackwall-xdp/src/manager.rs @@ -94,20 +94,58 @@ struct MirrorOp { impl MirrorOp { /// The identity this mirror op concerns, for coalescing purposes. fn key(&self) -> MirrorKey { - match self.action { - XdpAction::RateLimit { src, .. } | XdpAction::ClearRate { src } => MirrorKey::Src(src), - XdpAction::Block { net } | XdpAction::Unblock { net } => MirrorKey::Net(net), - } + mirror_key_of(&self.action) } } -/// The identity a queued mirror op is coalesced on. +/// The identity a queued mirror/reapply op is coalesced on. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum MirrorKey { Src(IpAddr), Net(IpNet), } +/// The coalescing identity of `action` — shared by [`MirrorOp::key`] and +/// [`ReapplyOp::key`]. +fn mirror_key_of(action: &XdpAction) -> MirrorKey { + match *action { + XdpAction::RateLimit { src, .. } | XdpAction::ClearRate { src } => MirrorKey::Src(src), + XdpAction::Block { net } | XdpAction::Unblock { net } => MirrorKey::Net(net), + } +} + +/// A [`XdpManager::reapply_active`] re-apply that failed at the +/// [`XdpExecutor`] and is queued for a self-heal retry (issue #194). +/// +/// Unlike [`MirrorOp`] (which only ever replays a journal write, the +/// executor side already having succeeded), a queued `ReapplyOp` re-attempts +/// the executor `apply` itself — `reapply_active`'s failure happens on the +/// executor side, not the journal side (`reapply_active` never re-journals +/// in the first place). +/// +/// Holds only the coalescing [`MirrorKey`] identity, not the action that was +/// captured when the op was queued: [`XdpManager::retry_pending_reapply`] +/// re-derives the CURRENT action from [`XdpController::current_rate_limit`]/ +/// [`XdpController::current_block`] at retry time rather than replaying a +/// snapshot, so a fresh, successful re-apply of the same identity with +/// different parameters (e.g. an operator or detection tightening a +/// `RateLimit`'s `pps`) that lands between the failed reapply and the retry +/// is never clobbered by the stale queued content (#194 C1 follow-up). +/// Mirrors `blackwall_rtbh::manager::RtbhManager`'s private `ReapplyOp`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ReapplyOp { + /// The identity of the action to re-apply; the current action content + /// is looked up fresh from the controller at retry time. + key: MirrorKey, +} + +impl ReapplyOp { + /// The identity this reapply op concerns, for coalescing purposes. + fn key(&self) -> MirrorKey { + self.key + } +} + /// Outcome of [`XdpManager::execute_and_journal`]. /// /// The auto path (`on_detection`, which always passes `fresh = true`) @@ -157,6 +195,16 @@ pub struct XdpManager { /// succeeded; retried (never re-issued to the executor) by /// [`XdpManager::retry_pending_mirror`] on the next tick. pending_mirror: Vec, + /// [`Self::reapply_active`] re-applies that failed at the + /// [`XdpExecutor`]; retried by [`Self::retry_pending_reapply`] on the + /// next tick (issue #194). The controller's active entry from + /// [`XdpController::mark_resumed`] is kept (never rolled back) while + /// queued — a re-installed entry is a known-good persisted mitigation + /// with no fresh detection to naturally re-attempt it, so rollback would + /// strand the control plane believing nothing is installed while the + /// journal still says otherwise (mirrors + /// `blackwall_rtbh::manager::RtbhManager`'s private `pending_reapply`). + pending_reapply: Vec, /// Count of executor applies that failed, each counted here (see /// [`Self::apply_failures`]); a fresh insert among them is also rolled /// back (see [`XdpController::rollback`]). @@ -182,6 +230,7 @@ impl XdpManager { executor, journal, pending_mirror: Vec::new(), + pending_reapply: Vec::new(), apply_failures: 0, disarmed: false, disarmed_skips: 0, @@ -297,11 +346,16 @@ impl XdpManager { /// Drain-retry any journal mirror writes queued by a previous transient /// failure. Call periodically. /// - /// The executor side of each queued op already succeeded when it was - /// queued, so this only ever re-attempts the matching journal call — it - /// never re-applies to the executor. + /// The executor side of each queued mirror op already succeeded when it + /// was queued, so that retry only ever re-attempts the matching journal + /// call — it never re-applies to the executor. Then retries any + /// `reapply_active` re-applies queued by a previous tick's transient + /// executor failure (see [`Self::retry_pending_reapply`], issue #194), + /// so both self-heals converge within one tick interval of the + /// respective dependency recovering. pub async fn tick(&mut self) { self.retry_pending_mirror().await; + self.retry_pending_reapply().await; } /// Re-install persisted active entries on a fresh session (rehydration). @@ -311,12 +365,18 @@ impl XdpManager { /// but does **not** re-journal, since the row already exists in the /// journal. An executor failure here is logged; the entry is still kept /// in the controller's active set (matching `RtbhManager::rehydrate`'s - /// "never silently drop a persisted row" invariant). + /// "never silently drop a persisted row" invariant) and queued via + /// [`Self::queue_reapply`] for a retry on the next [`Self::tick`] (issue + /// #194): unlike a live detection, a re-installed entry has no natural + /// re-detection to compensate for a dropped map write. pub async fn reapply_active(&mut self, rows: Vec<(XdpAction, XdpOrigin)>) { for (action, origin) in rows { self.controller.mark_resumed(&action, origin); if let Err(e) = self.executor.apply(action).await { - tracing::warn!(error = %e, ?action, "XDP: reapply_active executor call failed"); + tracing::warn!(error = %e, ?action, "XDP: reapply_active executor call failed; queuing for retry"); + self.queue_reapply(ReapplyOp { + key: mirror_key_of(&action), + }); } } } @@ -347,6 +407,18 @@ impl XdpManager { self.apply_failures } + /// Number of [`Self::reapply_active`] re-applies currently queued for a + /// self-heal retry after a failed executor apply on restart (issue + /// #194). Each queued entry is still active in the controller (kept, not + /// rolled back) but not yet confirmed on the map; drained by + /// [`Self::retry_pending_reapply`] on the next [`Self::tick`]. Surfaced + /// for `/metrics` as `blackwall_xdp_reapply_pending`, mirroring + /// `blackwall_rtbh::manager::RtbhManager::reapply_pending`. + #[must_use] + pub fn reapply_pending(&self) -> usize { + self.pending_reapply.len() + } + /// In-daemon disarm kill switch (C5): withdraw every currently-active /// block/rate-limit and switch to record-only for the rest of this /// process's life. @@ -412,6 +484,15 @@ impl XdpManager { self.pending_mirror.push(op); } + /// Queue a failed [`Self::reapply_active`] re-apply for retry, coalescing + /// by identity (source or network), same as [`Self::queue_mirror`] + /// (issue #194). + fn queue_reapply(&mut self, op: ReapplyOp) { + let key = op.key(); + self.pending_reapply.retain(|o| o.key() != key); + self.pending_reapply.push(op); + } + /// Execute one controller action on the executor and mirror it into the journal. /// /// `fresh` marks whether `action` is a brand-new insert (a first-time @@ -489,6 +570,44 @@ impl XdpManager { } } + /// Drain-retry queued [`Self::reapply_active`] re-applies left over from + /// a transient executor failure (issue #194). + /// + /// Each queued op re-derives the CURRENT action for its identity from + /// [`XdpController::current_rate_limit`]/[`XdpController::current_block`] + /// rather than replaying the action snapshot captured when the op was + /// queued: if the identity is no longer active (e.g. cleared by a manual + /// remove between the failed reapply and this tick), the lookup returns + /// `None` and the op is dropped — re-applying an entry the control plane + /// no longer wants live would itself create a phantom. If the identity + /// is still active but a fresh, successful re-apply changed its + /// parameters in the meantime (e.g. tightening a `RateLimit`'s `pps`), + /// re-deriving picks up that CURRENT action instead of replaying the + /// stale queued one, which would otherwise silently revert the fresh + /// update (#194 C1 follow-up). Otherwise the apply is re-attempted; ops + /// that still fail are kept (retried again on the next call), ops that + /// succeed are dropped. + async fn retry_pending_reapply(&mut self) { + if self.pending_reapply.is_empty() { + return; + } + let ops = std::mem::take(&mut self.pending_reapply); + for op in ops { + let current = match op.key { + MirrorKey::Src(src) => self.controller.current_rate_limit(src), + MirrorKey::Net(net) => self.controller.current_block(net), + }; + let Some(action) = current else { + tracing::info!(key = ?op.key, "XDP: dropping queued reapply; entry no longer active"); + continue; + }; + if let Err(e) = self.executor.apply(action).await { + tracing::warn!(error = %e, ?action, "XDP: reapply retry failed; re-queuing"); + self.pending_reapply.push(op); + } + } + } + #[cfg(test)] pub(crate) fn executor(&self) -> &E { &self.executor @@ -510,17 +629,31 @@ impl XdpManager { mod tests { use super::*; use blackwall_flow::{AttackKind, Detection, DetectionEvent, Severity}; - use std::sync::Mutex; + use std::sync::{Arc, Mutex}; - #[derive(Default)] + #[derive(Default, Clone)] struct FakeExecutor { - applied: Mutex>, - fail: bool, + applied: Arc>>, + fail: Arc>, /// If `Some(n)`, the n-th call (1-indexed) onward fails; earlier /// calls succeed. Used to simulate a successful fresh insert /// followed by a failing upgrade apply (Fix 2 regression test). fail_from_call: Option, - call_count: Mutex, + call_count: Arc>, + } + impl FakeExecutor { + /// Build a fake whose `apply` fails from the start. + fn with_fail(fail: bool) -> Self { + let f = Self::default(); + f.set_fail(fail); + f + } + /// Flip the failure toggle at runtime — lets a test simulate the + /// executor recovering mid-scenario (a clone shares the same + /// underlying flag with whatever manager holds this fake). + fn set_fail(&self, fail: bool) { + *self.fail.lock().unwrap() = fail; + } } #[async_trait] impl XdpExecutor for FakeExecutor { @@ -530,7 +663,8 @@ mod tests { *count += 1; *count }; - if self.fail || self.fail_from_call.is_some_and(|from| call_no >= from) { + if *self.fail.lock().unwrap() || self.fail_from_call.is_some_and(|from| call_no >= from) + { return Err(XdpExecError); } self.applied.lock().unwrap().push(action); @@ -600,10 +734,7 @@ mod tests { fn mgr(fail_exec: bool, fail_journal: bool) -> XdpManager { XdpManager::new( XdpController::new(own(), 100, 1000, Vec::new()), - FakeExecutor { - fail: fail_exec, - ..Default::default() - }, + FakeExecutor::with_fail(fail_exec), FakeJournal { fail: fail_journal, ..Default::default() @@ -881,6 +1012,61 @@ mod tests { assert_eq!(m.disarmed_skips(), 1); } + #[tokio::test] + async fn disarm_tolerates_a_withdraw_error() { + // Best-effort: an executor Err on disarm's withdraw (ClearRate) apply + // must not abort the sweep (a second active entry is still swept) or + // stop the manager from switching to record-only. Mirrors + // `blackwall_rtbh::manager::RtbhManager`'s + // `disarm_tolerates_a_withdraw_error`. + let mut m = XdpManager::new( + XdpController::new(own(), 100, 1000, Vec::new()), + FakeExecutor { + // Calls 1-2 (the two fresh rate-limit installs) succeed; + // call 3 onward (disarm's ClearRate withdraws) fail. + fail_from_call: Some(3), + ..Default::default() + }, + FakeJournal::default(), + ); + m.on_detection( + &DetectionEvent::Opened(det("203.0.113.7", vec!["198.51.100.9"])), + 0, + ) + .await; + m.on_detection( + &DetectionEvent::Opened(det("203.0.113.8", vec!["198.51.100.10"])), + 0, + ) + .await; + assert_eq!(m.active().len(), 2); + + m.disarm(1_000).await; + + assert!( + !m.executor() + .applied + .lock() + .unwrap() + .iter() + .any(|a| matches!(a, XdpAction::ClearRate { .. })), + "every withdraw errored, so no ClearRate was recorded by the fake" + ); + assert!( + m.active().is_empty(), + "disarm clears the active set even when every withdraw errors (best-effort)" + ); + + // Record-only holds even though disarm itself never got a + // confirmed withdraw. + m.on_detection( + &DetectionEvent::Opened(det("203.0.113.9", vec!["198.51.100.11"])), + 2_000, + ) + .await; + assert!(m.active().is_empty()); + } + #[tokio::test] async fn apply_add_while_disarmed_is_rejected_not_applied() { // C5 + final-review fix: a manual add while disarmed must be @@ -1005,4 +1191,123 @@ mod tests { assert!(journal.record(&block, XdpOrigin::Manual, 0).await.is_ok()); assert!(journal.record(&rate, XdpOrigin::Auto, 1000).await.is_ok()); } + + #[tokio::test] + async fn reapply_active_failure_queues_reapply_and_tick_reconverges() { + // #194 C1: a persisted, active entry whose reapply fails at the + // executor must NOT be left stranded (active in the controller but + // never written to the eBPF map) — it is queued for retry and the + // queue drains once the executor recovers. + let executor = FakeExecutor::with_fail(true); // apply errors + let mut m = XdpManager::new( + XdpController::new(own(), 100, 1000, Vec::new()), + executor.clone(), + FakeJournal::default(), + ); + let net: IpNet = "198.51.100.0/24".parse().unwrap(); + let action = XdpAction::Block { net }; + m.reapply_active(vec![(action, XdpOrigin::Manual)]).await; + assert!( + m.active().iter().any(|(a, _)| *a == action), + "entry kept (not dropped)" + ); + assert_eq!(m.reapply_pending(), 1, "failed reapply queued for retry"); + + // Executor recovers; the next tick re-applies and drains the queue. + executor.set_fail(false); + m.tick().await; + assert_eq!(m.reapply_pending(), 0); + assert!(m.executor().applied.lock().unwrap().contains(&action)); + } + + #[tokio::test] + async fn queue_reapply_dedupes_and_drops_if_cleared() { + // A still-failing tick must coalesce (not double-enqueue) the retry + // for the same entry; and once the entry is cleared before it ever + // succeeds, the queued reapply must be dropped rather than + // re-applying a no-longer-wanted action. + let executor = FakeExecutor::with_fail(true); + let mut m = XdpManager::new( + XdpController::new(own(), 100, 1000, Vec::new()), + executor.clone(), + FakeJournal::default(), + ); + let net: IpNet = "198.51.100.0/24".parse().unwrap(); + let action = XdpAction::Block { net }; + m.reapply_active(vec![(action, XdpOrigin::Manual)]).await; + m.tick().await; // still failing -> re-queued, NOT double-enqueued + assert_eq!(m.reapply_pending(), 1, "coalesced, not doubled"); + + // Cleared before it ever succeeded (manual unblock). + m.apply_remove(net, 3_000).await; + executor.set_fail(false); + m.tick().await; + assert_eq!(m.reapply_pending(), 0, "dropped: entry no longer active"); + assert!( + !m.executor().applied.lock().unwrap().contains(&action), + "not re-applied after clear" + ); + } + + #[tokio::test] + async fn stale_reapply_does_not_clobber_a_fresh_successful_update() { + // #194 C1 follow-up: `retry_pending_reapply` must re-derive the + // controller's CURRENT action for the identity at retry time, not + // replay the action snapshot captured when the op was queued. + // Otherwise a fresh, successful re-apply of the SAME source with + // DIFFERENT parameters that lands between the failed reapply and + // the retry gets silently reverted by the stale queued content — a + // mitigation-weakening regression (e.g. a fresh `pps:1000` reverted + // back to a stale queued `pps:500`). + let executor = FakeExecutor::with_fail(true); // reapply_active apply fails + let mut m = XdpManager::new( + XdpController::new(own(), 100, 1000, Vec::new()), + executor.clone(), + FakeJournal::default(), + ); + let addr: IpAddr = "198.51.100.9".parse().unwrap(); + let old = XdpAction::RateLimit { + src: addr, + pps: 500, + burst: 500, + victim: None, + }; + m.reapply_active(vec![(old, XdpOrigin::Manual)]).await; + assert!( + m.active().iter().any(|(a, _)| *a == old), + "entry kept (not dropped)" + ); + assert_eq!(m.reapply_pending(), 1, "failed reapply queued for retry"); + + // Executor recovers just in time for a FRESH, successful re-apply + // with DIFFERENT parameters for the SAME source, landing before the + // next tick drains the stale queue. + executor.set_fail(false); + let outcome = m.apply_rate_limit(addr, 1000, 1000, 1_500).await; + assert_eq!(outcome, ApplyOutcome::Applied); + let new = XdpAction::RateLimit { + src: addr, + pps: 1000, + burst: 1000, + victim: None, + }; + assert!( + m.executor().applied.lock().unwrap().contains(&new), + "the fresh update reached the executor" + ); + + // The next tick drains the (now-stale) queued reapply. It must + // re-derive and re-apply the CURRENT action (pps 1000), not replay + // the stale queued snapshot (pps 500) — which would silently revert + // the fresh update. + m.tick().await; + assert_eq!(m.reapply_pending(), 0); + + let applied = m.executor().applied.lock().unwrap(); + let last = applied.last().expect("at least one apply recorded"); + assert_eq!( + *last, new, + "the drained retry must re-apply the CURRENT action, not the stale queued one" + ); + } } diff --git a/scripts/coverage.sh b/scripts/coverage.sh index 3c518d0..0294e51 100755 --- a/scripts/coverage.sh +++ b/scripts/coverage.sh @@ -15,6 +15,7 @@ # - blackwall-xdp/src/dataplane.rs aya load/attach + live eBPF map I/O (needs CAP_NET_ADMIN + kernel) # - blackwall-xdp/src/afxdp.rs AF_XDP socket/UMEM/ring I/O (needs CAP_NET_ADMIN/RAW + a live iface) # - blackwall-xdp/src/capture.rs capture-ring drain + flag toggle over pinned bpffs maps (live kernel) +# - blackwall-rpki/src/fetch.rs RPKI validator HTTP GET (reqwest) + the check-pass loop over it # Every one of these is a thin adapter; all of its non-trivial pure logic lives # in unit-tested helpers (e.g. transport/packet.rs, render.rs, *_parse.rs). # @@ -29,6 +30,6 @@ # Extra args are forwarded to cargo llvm-cov (e.g. --html, --summary-only). set -euo pipefail -EXCLUDE='(transport/(tproxy|nfqueue)\.rs|blackwall-nft/src/apply\.rs|blackwalld/src/(main|metrics|api|shadow)\.rs|discovery/src/incus_client\.rs|discovery/src/proc_io\.rs|speedtest/src/providers/.*_net\.rs|shaper/src/apply\.rs|dns/src/send_net\.rs|flow/src/collector_net\.rs|bgp/src/session_net\.rs|blackwall-lab/src/exec/.*\.rs|blackwall-lab/src/cli\.rs|blackwall-lab/src/bin/lab\.rs|blackwall-trafficgen/src/io/.*\.rs|blackwall-trafficgen/src/bin/.*\.rs|blackwall-xdp/src/dataplane\.rs|blackwall-xdp/src/afxdp\.rs|blackwall-xdp/src/capture\.rs)' +EXCLUDE='(transport/(tproxy|nfqueue)\.rs|blackwall-nft/src/apply\.rs|blackwalld/src/(main|metrics|api|shadow)\.rs|discovery/src/incus_client\.rs|discovery/src/proc_io\.rs|speedtest/src/providers/.*_net\.rs|shaper/src/apply\.rs|dns/src/send_net\.rs|flow/src/collector_net\.rs|bgp/src/session_net\.rs|blackwall-lab/src/exec/.*\.rs|blackwall-lab/src/cli\.rs|blackwall-lab/src/bin/lab\.rs|blackwall-trafficgen/src/io/.*\.rs|blackwall-trafficgen/src/bin/.*\.rs|blackwall-xdp/src/dataplane\.rs|blackwall-xdp/src/afxdp\.rs|blackwall-xdp/src/capture\.rs|blackwall-rpki/src/fetch\.rs)' exec cargo llvm-cov --workspace --fail-under-lines 90 --ignore-filename-regex "$EXCLUDE" "$@"