Pre-M0 detection-quality batch (D1–D7, S1–S3)

CHANGELOG.md

@@ -9,6 +9,7 @@ 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
+- 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/<pkg>-<test>`) instead of `cargo test`, removing the cargo lock/daemon from the scenario entirely.
 - RTBH hardening ahead of production wiring (sub-project C, C1c Part A) — fixes a code review found in the C1b controller and C1a speaker. **The RTBH controller leaked blackholes permanently:** the flow detector emits `Cleared` once, and if it arrived before the RTBH `hold_down` elapsed the withdraw was dropped with no retry, so the victim `/32` stayed blackholed (self-DoSed) forever. The controller now defers such a clear and completes it from a periodic `tick`, adds a `max_ttl` auto-blackhole backstop against a dropped/missed `Cleared`, tracks per-entry origin so an operator (`Manual`) blackhole is never auto-cleared, and cancels a deferred clear when a re-attack (`Opened`/`Updated`) re-asserts the target (`blackwall-rtbh`). **The BGP speaker could panic on a large config:** `push_attr` truncated attribute lengths to one byte, so ≥64 communities overflowed and killed the session task silently — it now emits RFC 4271 §4.3 extended-length attributes; `encode_nlri` also truncates host bits so a non-host prefix can't emit malformed NLRI (`blackwall-bgp`). The speaker further validates its config as iBGP (`local_asn == peer_asn`) and the peer's OPEN (ASN and hold-time, rejecting with the correct NOTIFICATION codes), waits for the peer's KEEPALIVE before declaring the session Established (bounded by the OpenConfirm hold timer), and surfaces send failures on `BgpHandle` instead of silently dropping routes. Controller logic is unit-tested; the speaker changes are covered by the `bgp-bird` lab gate.
 - sFlow decoder dropped every flow sample from real agents: it handled only regular flow samples (type 1), but hsflowd and many switches emit **expanded flow samples (type 3)**, so `decode_datagram` returned zero observations and the volumetric detector never fired on real-agent traffic. The decoder now handles expanded flow samples (`blackwall-flow`); covered by byte-exact tests over real captured hsflowd datagrams. Found by the new `flow-sflow-live` lab gate.

bin/blackwalld/src/main.rs

@@ -89,7 +89,9 @@ enum Command {
         /// Per-destination packets-per-second threshold.
         #[arg(long)]
         pps_threshold: f64,
-        /// Per-destination bits-per-second threshold.
+        /// Per-destination bits-per-second threshold. NOTE: computed from the sFlow
+        /// L2 frame length (includes the Ethernet header), so calibrate against L2
+        /// bytes, not L3 payload.
         #[arg(long)]
         bps_threshold: f64,
         /// Sliding window in seconds.
@@ -98,6 +100,18 @@ enum Command {
         /// Hold-down in seconds before clearing a detection.
         #[arg(long, default_value_t = 30)]
         hold_down_secs: u64,
+        /// Minimum raw sFlow samples in-window before a detection may open (guards
+        /// sampling-variance false positives). 0 disables the gate.
+        #[arg(long, default_value_t = 8)]
+        min_samples: usize,
+        /// Ceiling multiplier applied to an agent's expected sampling rate when
+        /// its reported rate is high. A reported rate above `expected * 4` is
+        /// trusted (adaptive samplers legitimately raise their rate under load)
+        /// up to `expected * max_sampling_factor`, and only clamped down beyond
+        /// that ceiling — never clamped down to `expected` itself, which would
+        /// mask a real flood as a false negative.
+        #[arg(long, default_value_t = 64)]
+        max_sampling_factor: u32,
     },
     /// Apply the ruleset and start the deception engine (requires CAP_NET_ADMIN).
     Run {
@@ -1298,7 +1312,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             Ok(())
         }
         Command::BirdConfig { config } => {
-            let policy = blackwall_config::parse_file(&config)?;
+            let policy = blackwall_config::parse_and_resolve(&config)?;
             match blackwall_bgp::render_bird_ibgp(&policy) {
                 Ok(s) => {
                     print!("{s}");
@@ -1353,8 +1367,10 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             bps_threshold,
             window_secs,
             hold_down_secs,
+            min_samples,
+            max_sampling_factor,
         } => {
-            let policy = blackwall_config::parse_file(&config)?;
+            let policy = blackwall_config::parse_and_resolve(&config)?;
             if policy.shadow {
                 tracing::warn!(
                     "SHADOW MODE — mitigations are LOGGED, NOT APPLIED (RTBH/FlowSpec/XDP)"
@@ -1365,12 +1381,17 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
             let store = std::sync::Arc::new(blackwall_state::Store::connect(&database_url).await?);
             store.migrate().await?;
             let agents = blackwall_flow::AgentRegistry::from_entries(&policy.pops);
-            let detector = blackwall_flow::ThresholdDetector::new(
-                policy.prefixes.clone(),
+            let detector_config = blackwall_flow::DetectorConfig {
                 pps_threshold,
                 bps_threshold,
-                window_secs * 1000,
-                hold_down_secs * 1000,
+                window_ms: window_secs * 1000,
+                hold_down_ms: hold_down_secs * 1000,
+                min_samples,
+                max_sampling_factor,
+            };
+            let detector = blackwall_flow::ThresholdDetector::new(
+                policy.prefixes.clone(),
+                detector_config,
                 agents,
             );
 

bin/blackwalld/src/metrics.rs

@@ -100,6 +100,12 @@ async fn gather(sources: &MetricsSources) -> Vec<Metric> {
             kind: MetricKind::Counter,
             value: u64_to_f64(collector.decode_errors()),
         });
+        m.push(Metric {
+            name: "blackwall_flow_sample_decode_errors_total",
+            help: "sFlow samples that failed to decode within an otherwise-valid datagram",
+            kind: MetricKind::Counter,
+            value: u64_to_f64(collector.sample_decode_errors()),
+        });
     }
     if let Some(inflight) = &sources.inflight {
         m.push(Metric {
@@ -240,8 +246,10 @@ fn xdp_block(sources: &MetricsSources) -> Option<String> {
 }
 
 /// Render the per-POP telemetry blocks (`blackwall_flow_pop_last_seen_seconds`,
-/// `blackwall_flow_agent_sampling_mismatch_total`) plus the
-/// `blackwall_flow_unknown_agent_observations_total` scalar, or `None` when the
+/// `blackwall_flow_agent_sampling_mismatch_total`,
+/// `blackwall_flow_sampling_near_ceiling_total`) plus the
+/// `blackwall_flow_unknown_agent_observations_total` and
+/// `blackwall_flow_min_sample_suppressed_total` scalars, or `None` when the
 /// flow daemon has no per-agent snapshot wired up (`sources.agent_stats` is
 /// `None` — the deception engine, which has no sFlow collector).
 ///
@@ -289,6 +297,21 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
                 s.pop, s.mismatches
             );
         }
+        let _ = writeln!(
+            out,
+            "\n# HELP blackwall_flow_sampling_near_ceiling_total Samples per POP whose trusted sampling rate landed at or above half the max-sampling-factor ceiling"
+        );
+        let _ = writeln!(
+            out,
+            "# TYPE blackwall_flow_sampling_near_ceiling_total counter"
+        );
+        for s in &stats {
+            let _ = writeln!(
+                out,
+                "blackwall_flow_sampling_near_ceiling_total{{pop=\"{}\"}} {}",
+                s.pop, s.near_ceiling
+            );
+        }
     }
 
     // Always emitted (a single scalar, not per-label) so the series exists
@@ -313,6 +336,61 @@ fn agent_stats_block(sources: &MetricsSources, now_ms: u64) -> Option<String> {
         "blackwall_flow_unknown_agent_observations_total {unknown}"
     );
 
+    // Always emitted alongside `unknown_agent_observations_total`, same reasoning.
+    let min_sample_suppressed = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.min_sample_suppressed());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_min_sample_suppressed_total Detections suppressed by the minimum-sample gate"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_min_sample_suppressed_total counter"
+    );
+    let _ = writeln!(
+        out,
+        "blackwall_flow_min_sample_suppressed_total {min_sample_suppressed}"
+    );
+
+    // Always emitted alongside the other scalars above, same reasoning:
+    // distinguishes "quiet network" (both zero) from "POP silently dropping
+    // samples" (decode/suppression counters moving but detections never open).
+    let detections_opened = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.detections_opened());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_detections_opened_total Detections opened by the flow detector"
+    );
+    let _ = writeln!(out, "# TYPE blackwall_flow_detections_opened_total counter");
+    let _ = writeln!(
+        out,
+        "blackwall_flow_detections_opened_total {detections_opened}"
+    );
+
+    let detections_cleared = sources
+        .collector
+        .as_ref()
+        .map_or(0, |c| c.detections_cleared());
+    out.push('\n');
+    let _ = writeln!(
+        out,
+        "# HELP blackwall_flow_detections_cleared_total Detections cleared by the flow detector"
+    );
+    let _ = writeln!(
+        out,
+        "# TYPE blackwall_flow_detections_cleared_total counter"
+    );
+    let _ = writeln!(
+        out,
+        "blackwall_flow_detections_cleared_total {detections_cleared}"
+    );
+
     Some(out)
 }
 

crates/blackwall-api/src/handlers.rs

@@ -14,6 +14,20 @@ use std::sync::Arc;
 /// Default row cap for the `sessions`/`audit` feeds.
 const DEFAULT_LIMIT: i64 = 100;
 
+/// Upper bound on the `sessions`/`audit` feed `?limit=`, regardless of what
+/// the caller requests.
+const MAX_LIMIT: i64 = 1000;
+
+/// Clamp a requested feed `limit` to `[1, MAX_LIMIT]`, defaulting to
+/// `DEFAULT_LIMIT` when absent.
+///
+/// Guards the `LIMIT $1` bind in the feed queries: a negative or zero value
+/// would otherwise reach Postgres and error out, and an unbounded value
+/// would dump the entire table.
+fn clamp_limit(limit: Option<i64>) -> i64 {
+    limit.unwrap_or(DEFAULT_LIMIT).clamp(1, MAX_LIMIT)
+}
+
 /// `?limit=` query for the capped feeds.
 #[derive(Debug, Deserialize)]
 pub struct LimitQuery {
@@ -178,7 +192,7 @@ pub async fn list_sessions(
     State(s): St,
     Query(q): Query<LimitQuery>,
 ) -> ApiResult<Json<Vec<SessionDto>>> {
-    let limit = q.limit.unwrap_or(DEFAULT_LIMIT);
+    let limit = clamp_limit(q.limit);
     Ok(Json(
         s.sessions(limit)
             .await?
@@ -200,7 +214,7 @@ pub async fn list_audit(
     State(s): St,
     Query(q): Query<LimitQuery>,
 ) -> ApiResult<Json<Vec<AuditDto>>> {
-    let limit = q.limit.unwrap_or(DEFAULT_LIMIT);
+    let limit = clamp_limit(q.limit);
     Ok(Json(
         s.audit(limit)
             .await?
@@ -209,3 +223,17 @@ pub async fn list_audit(
             .collect(),
     ))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn limit_is_clamped() {
+        assert_eq!(clamp_limit(Some(-1)), 1);
+        assert_eq!(clamp_limit(Some(0)), 1);
+        assert_eq!(clamp_limit(Some(i64::MAX)), MAX_LIMIT);
+        assert_eq!(clamp_limit(None), DEFAULT_LIMIT);
+        assert_eq!(clamp_limit(Some(50)), 50);
+    }
+}

crates/blackwall-config/src/error.rs

@@ -34,4 +34,12 @@ pub enum ConfigError {
     /// The config file could not be read.
     #[error("reading config: {0}")]
     Io(#[from] std::io::Error),
+    /// The config parsed successfully but is semantically invalid: an owned
+    /// address falls outside every managed prefix, two tenants claim the
+    /// same address, or the same service is defined more than once. Surfaced
+    /// by [`crate::parse_and_resolve`] so load-time paths (`flow`,
+    /// `bird-config`) fail before ever running, instead of misbehaving
+    /// silently at apply time.
+    #[error("config failed validation: {0}")]
+    Resolve(#[from] blackwall_core::PolicyError),
 }

crates/blackwall-config/src/lexer.rs

@@ -24,7 +24,11 @@ pub fn lex(input: &str) -> Vec<Line> {
     let mut pending: Option<(usize, Vec<String>)> = None;
 
     for (idx, raw) in input.lines().enumerate() {
-        let without_comment = match raw.find('#') {
+        let comment_start = raw
+            .char_indices()
+            .find(|&(i, c)| c == '#' && (i == 0 || raw[..i].ends_with(|p: char| p.is_whitespace())))
+            .map(|(i, _)| i);
+        let without_comment = match comment_start {
             Some(pos) => &raw[..pos],
             None => raw,
         };
@@ -110,4 +114,23 @@ mod tests {
         assert_eq!(lines.len(), 1);
         assert_eq!(lines[0].words, vec!["interface", "wan", "eth0"]);
     }
+
+    #[test]
+    fn hash_inside_token_is_literal() {
+        let out = lex("rtbh md5=sec#ret\n");
+        assert_eq!(out[0].words, vec!["rtbh", "md5=sec#ret"]);
+    }
+
+    #[test]
+    fn key_equals_hash_value_is_literal() {
+        let out = lex("k key=#val\n");
+        assert_eq!(out[0].words, vec!["k", "key=#val"]);
+    }
+
+    #[test]
+    fn whitespace_preceded_hash_still_comments() {
+        assert_eq!(lex("foo # comment\n")[0].words, vec!["foo"]);
+        assert!(lex("# comment only\n").is_empty());
+        assert!(lex("    # indented comment\n").is_empty());
+    }
 }

crates/blackwall-config/src/lib.rs

@@ -20,10 +20,29 @@ pub fn parse_file(path: &Path) -> Result<Policy, ConfigError> {
     parse_str(&text)
 }
 
+/// Parse a config file and validate it (`Policy::resolve()`), so a
+/// semantically invalid config (address outside prefixes, duplicate
+/// ownership, duplicate service) fails at load rather than at apply. Used by
+/// load-time paths (`flow`, `bird-config`).
+pub fn parse_and_resolve(path: &Path) -> Result<Policy, ConfigError> {
+    let policy = parse_file(path)?;
+    policy.resolve().map_err(ConfigError::Resolve)?;
+    Ok(policy)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    /// Test-only sibling of [`parse_and_resolve`] that parses from an
+    /// in-memory string instead of a file, mirroring its parse-then-resolve
+    /// behavior for unit tests that don't want to touch the filesystem.
+    fn parse_and_resolve_str(input: &str) -> Result<Policy, ConfigError> {
+        let policy = parse_str(input)?;
+        policy.resolve().map_err(ConfigError::Resolve)?;
+        Ok(policy)
+    }
+
     #[test]
     fn parse_str_round_trips_through_resolve() {
         let policy = parse_str(
@@ -35,4 +54,17 @@ mod tests {
         assert_eq!(resolved.len(), 1);
         assert_eq!(resolved[0].port, 443);
     }
+
+    #[test]
+    fn parse_and_resolve_rejects_out_of_prefix_ownership() {
+        let cfg = "interface wan eth0\nipv4 203.0.113.0/24\ntenant t {\n owns 198.51.100.7\n}\n";
+        let err = parse_and_resolve_str(cfg).unwrap_err();
+        assert!(matches!(err, ConfigError::Resolve(_)));
+    }
+
+    #[test]
+    fn parse_and_resolve_ok_for_flow_only_config() {
+        let cfg = "interface wan eth0\nipv4 203.0.113.0/24\nshadow\n";
+        assert!(parse_and_resolve_str(cfg).is_ok());
+    }
 }