From d2231a113c3db4b53678398b611352116eff2036 Mon Sep 17 00:00:00 2001 From: v0l Date: Sun, 12 Jul 2026 22:14:28 +0100 Subject: [PATCH] feat(fw): move per-source rate limiting into the XDP datapath MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-source rate machine (counting, rate calculation, and the blocking decision) now lives entirely in the kernel. The eBPF datapath tracks each source over a fixed 1s window in V4/V6_SRC_STATE and blocks over-rate sources in-line (blocked_until, re-extended each window still over-rate); drops engage once the destination escalates to SOURCE_BLOCK. Userspace only arms SRC_RATE_CFG (startup + PUT /limits) and batch-reads the state for display. Deleted: cumulative V4/V6_SRC_COUNTERS, the auto V4/V6_CIDR_SRC block tries, SourceTracker/advance_source/step_sources, CIDR aggregation (plan_blocks), and the spoof gate. The seeding/counter-lifecycle bug class is structurally gone; blocking latency is packet-exact instead of up-to-one-500ms-tick; control-loop CPU no longer scales with flood history. Also: raw BPF_MAP_LOOKUP_BATCH reader (batch.rs) for all remaining map scans (dest/tx counters, GC sweeps, state snapshots) with per-entry fallback on old kernels — one syscall per ~4k entries instead of two per entry; idle source states GC'd after 60s. Compat: legacy escalation config keys (agg-fanout, src-exit-pct, spoof gate, …) still parse and are ignored (regression-tested) so existing config files keep loading across self-upgrades. src_exit_pct is removed from the Limits API; the dashboard drops the cooling state and src exit field. Verified: 43 lib + 2 bin + 15 API tests and the full netns e2e suite (21 root tests) pass, including two new scenarios proving the kernel blocks and releases an over-rate source with zero userspace ticks. Dest/prefix detection deliberately stays in userspace: bounded (16k), multi-axis policy with prefix aggregation, and mitigation state persists in DEST_STATE regardless of the daemon. --- docs/agents/fw-api.md | 41 +- lnvps_fw/README.md | 29 +- lnvps_fw/lnvps_ebpf/src/main.rs | 13 +- lnvps_fw/lnvps_ebpf/src/maps.rs | 110 +++-- lnvps_fw/lnvps_fw_common/src/lib.rs | 33 ++ lnvps_fw/lnvps_fw_service/config.example.yaml | 54 +-- .../dashboard/dist/index.html | 2 +- .../lnvps_fw_service/dashboard/src/api.ts | 4 +- .../lnvps_fw_service/dashboard/src/cards.tsx | 5 +- .../lnvps_fw_service/examples/serve_api.rs | 1 - lnvps_fw/lnvps_fw_service/src/api.rs | 18 +- lnvps_fw/lnvps_fw_service/src/batch.rs | 324 +++++++++++++ lnvps_fw/lnvps_fw_service/src/cidr.rs | 391 +--------------- lnvps_fw/lnvps_fw_service/src/config.rs | 62 +-- lnvps_fw/lnvps_fw_service/src/detect.rs | 238 ---------- lnvps_fw/lnvps_fw_service/src/gc.rs | 50 +- lnvps_fw/lnvps_fw_service/src/lib.rs | 1 + lnvps_fw/lnvps_fw_service/src/main.rs | 186 +++++--- lnvps_fw/lnvps_fw_service/src/runtime.rs | 442 +++++++----------- lnvps_fw/lnvps_fw_service/tests/api.rs | 2 - .../lnvps_fw_service/tests/carpet_bomb.rs | 7 - lnvps_fw/lnvps_fw_service/tests/escalation.rs | 200 ++++---- .../lnvps_fw_service/tests/harness/mod.rs | 87 ++-- lnvps_fw/lnvps_fw_service/tests/mitigation.rs | 76 +-- work/fw-inkernel-src-rate.md | 98 ++++ 25 files changed, 1133 insertions(+), 1341 deletions(-) create mode 100644 lnvps_fw/lnvps_fw_service/src/batch.rs create mode 100644 work/fw-inkernel-src-rate.md diff --git a/docs/agents/fw-api.md b/docs/agents/fw-api.md index 58be7a70..5135a262 100644 --- a/docs/agents/fw-api.md +++ b/docs/agents/fw-api.md @@ -39,7 +39,7 @@ section to disable the API entirely. | GET | `/tracked` | live per-destination RX/TX rates (paginated + filtered) | | GET | `/ports` | learned open ports per protected IP (paginated + filtered) | | GET | `/sources` | **the** unified source list: every rate-tracked source (3-state + live pps) plus manual blocks, paginated + filtered | -| GET | `/blocks` | legacy: the enforced block trie only (manual + auto dropping/cooling, CIDR-aggregated), paginated | +| GET | `/blocks` | legacy: currently-blocked sources only (manual CIDRs + kernel-blocked /32s|/128s), paginated | | POST | `/blocks` | add a permanent manual source block `{cidr}` (updates the ruleset) | | DELETE | `/blocks?cidr=` | remove a manual source block | | GET | `/upgrade` | cached self-upgrade status: `current`, `latest`, `available`, `deb_url` | @@ -66,31 +66,38 @@ neither can hide: - `pps`/`syn_pps`/`bps` (+ `net_*` prefix aggregates) — *destination* entry thresholds: “is this dest under attack?” -- `src_rate_pps` (+ `src_exit_pct`, `src_cooldown_secs`) — the *per-source* - auto-block threshold: once a dest is mitigating, any single source at/over - this rate is blocked. Necessarily much lower than the dest threshold, but - keep it well above shared-infrastructure rates (CDN/reverse-proxy edges, - CGNAT) — default 10 000 pps. Mirrors `escalation.src-rate-pps` in the config - file; the API value wins after a PUT. +- `src_rate_pps` (+ `src_cooldown_secs`) — the *per-source* auto-block + threshold, enforced by the in-kernel rate machine over an exact 1s window: + once a dest is mitigating, any single source over this rate is blocked for + the cooldown. Necessarily much lower than the dest threshold, but keep it + well above shared-infrastructure rates (CDN/reverse-proxy edges, CGNAT) — + default 10 000 pps. Mirrors `escalation.src-rate-pps` in the config file; + the API value wins after a PUT (userspace re-writes the kernel config map). Omitting the `src_*` fields in a PUT (older clients) falls back to their defaults rather than zeroing them. ### Source list (`/sources`) -While a destination is under mitigation the datapath counts every source IP and -userspace runs a per-source rate state machine. `GET /sources` is the **single** -source list the UI shows: it surfaces that full set in every state, **plus** -operator-pushed manual blocks. There is no separate “blocks list” — an auto block -is simply an entry whose `state` is `dropping`/`cooling`. +While a destination is under mitigation, the **XDP datapath itself** runs a +fixed-window rate machine per source IP: it counts each source over a 1s +window and blocks over-rate sources in-line (`blocked_until` in its state +map), with drops engaging once the destination escalates to `SOURCE_BLOCK`. +Userspace is not in the decision path — it snapshots the kernel state map +(batched) purely for this view. `GET /sources` is the **single** source list +the UI shows, plus operator-pushed manual blocks. An auto block is simply an +entry whose `state` is `dropping`. Each item is `{ ip, pps, state, manual, age_secs }` where `state` is one of: -- `normal` — under the per-source limit (`src-rate-pps`); tracked but **not** blocked. -- `dropping` — at/over the limit; installed in the CIDR block trie (packets dropped). -- `cooling` — still blocked, but the rate fell below the exit threshold - (`src-exit-pct` of the limit) and is counting down `src-cooldown-secs` before - release. A new at/over-rate window flips it back to `dropping`. +- `normal` — under the per-source limit (`src-rate-pps`); counted but **not** + blocked. +- `dropping` — tripped the limit; the kernel drops its packets until the + cooldown expires (re-extended every window it stays over-rate). + +`pps` is estimated from the current kernel window (count over elapsed); an +idle source's rate decays toward zero naturally. A `normal` source idle for +60s is swept from the kernel state map by the daemon's GC timer. `manual: true` rows are permanent operator blocks (from `POST /blocks`); they are dropped before per-source counting so they always report `pps: 0` and are pinned diff --git a/lnvps_fw/README.md b/lnvps_fw/README.md index ff4c58d1..3741e816 100644 --- a/lnvps_fw/README.md +++ b/lnvps_fw/README.md @@ -39,9 +39,10 @@ single-IP attacks. high-efficacy open-port drop does the heavy lifting first; source/CIDR blocking (highest false-positive risk, useless against spoofed floods) is a last resort, and is gated so it never fires against spoofed traffic. -4. **Everything is bounded.** Per-source counters are LRU-bounded; the CIDR block - structure is an LPM trie kept small by aggregation. The datapath never tries - to track "every /32". +4. **Everything is bounded.** Per-source rate state is LRU-bounded and lives in + the datapath itself; a spoofed high-cardinality flood churns entries without + ever tripping a per-source limit. The daemon never scans unbounded state on + its hot path (reads are batched and display-only). --- @@ -71,8 +72,9 @@ single-IP attacks. ┌── lnvps_fw_service (userspace daemon) ───────────────────────┐ │ • samples per-dest + per-prefix counters, runs detection │ │ state machine, writes protection flags into DEST_STATE │ - │ • samples per-source counters, aggregates offenders into │ - │ CIDR blocks (spoof-gated), decays them │ + │ • arms the in-kernel per-source rate machine (which counts │ + │ and blocks over-rate sources in XDP on its own) and │ + │ snapshots its state for the API views │ │ • learned-port TTL GC, cookie-secret rotation, verified GC │ │ • loads config (interfaces, thresholds, protected prefixes) │ └──────────────────────────────────────────────────────────────┘ @@ -94,7 +96,7 @@ single-IP attacks. |---|---|---| | `lnvps_ebpf` | `bpfel-unknown-none` | The XDP + TC eBPF programs and maps. Not a default workspace member. | | `lnvps_fw_common` | host + eBPF (`#![no_std]`) | Types shared across the map boundary (counters, keys, `DestState`), the SYN-cookie function, and constants. `aya::Pod` impls behind the `user` feature. | -| `lnvps_fw_service` | host | The userspace daemon: config, detection state machine, CIDR aggregation, GC, and the control loop that programs the maps. Also exposes a library used by the test harness. | +| `lnvps_fw_service` | host | The userspace daemon: config, per-destination detection state machine, GC, control API, and the control loop that programs the maps (the per-source rate machine runs in the eBPF datapath). Also exposes a library used by the test harness. | --- @@ -133,8 +135,9 @@ cheaper layers have not already shed the attack: - **Spoofed SYN flood to an open TCP port**: `SYN_PROXY` answers with a SYN-cookie SYN-ACK; spoofed sources can't complete the handshake, so they never reach the protected host; real clients complete it and are allow-listed. -- **Real botnet hammering open services**: bounded real sources are aggregated - into CIDR blocks (`SOURCE_BLOCK`). +- **Real botnet hammering open services**: the in-kernel per-source rate + machine blocks each over-rate source the moment it crosses the limit + (`SOURCE_BLOCK` gates the drops; spoofed floods never trip it). --- @@ -166,7 +169,8 @@ the previous slot keeps in-flight cookies valid across a rotation. | `V4/V6_SRC_COUNTERS` | LRU per-CPU hash | XDP (under mitigation) | daemon (source control) | | `V4/V6_DEST_STATE` | LPM trie → `DestState` | daemon | XDP (mode lookup) | | `OPEN_PORTS_V4/V6` | LRU hash → `LastSeen` | TC egress (learning) | XDP; daemon (TTL GC) | -| `V4/V6_CIDR_SRC` | LPM trie | daemon (escalation) | XDP (`SOURCE_BLOCK`) | +| `V4/V6_SRC_STATE` | LRU hash | XDP (rate machine) | XDP (`SOURCE_BLOCK`) + daemon (display/GC) | +| `SRC_RATE_CFG` | array | daemon (limits) | XDP (rate machine) | | `VERIFIED_V4/V6` | LRU hash | XDP (`xdp_syn_proxy`/`_v6`) | XDP; daemon (TTL GC) | | `COOKIE_SECRET` | array[2] | daemon (rotation) | XDP | | `SYN_PROXY_JUMP` | prog array | daemon (setup) | XDP (`tail_call`) | @@ -292,8 +296,9 @@ keys, matching the rest of the LNVPS API config style. Key sections: - `thresholds` — per-destination entry thresholds + hysteresis/cooldown. - `network` — aggregate per-prefix thresholds (network scale). - `learning` — port-learning TTL and GC cadence. -- `escalation` — per-source rate, CIDR aggregation fan-out, spoof gate, and the - SYN-proxy trigger. +- `escalation` — per-source rate limit + cooldown (written into the in-kernel + rate machine), the SOURCE_BLOCK escalation gate, and the SYN-proxy trigger. + (Legacy aggregation/spoof-gate keys are still parsed but ignored.) > The `protected` prefixes and thresholds will be sourced from the LNVPS API in > a later increment; the local config is the bootstrap today. @@ -321,7 +326,7 @@ Test binaries (`lnvps_fw_service/tests/`): | `smoke` | program attach, per-dest counters, forwarding, SYN counting | | `learning` | TC egress learns open TCP/UDP ports; TTL GC | | `mitigation` | port-filter drops, learned-port pass, detect→cooldown, flag gating | -| `escalation` | per-source rate → aggregated CIDR block + decay | +| `escalation` | per-source rate limit for the in-kernel gate + SOURCE_BLOCK escalation | | `carpet_bomb` | thin prefix-wide flood flips the whole prefix | | `syn_proxy` | real client completes the cookie handshake + is verified; spoofed never verify | diff --git a/lnvps_fw/lnvps_ebpf/src/main.rs b/lnvps_fw/lnvps_ebpf/src/main.rs index 6ac5b58e..300e67e5 100644 --- a/lnvps_fw/lnvps_ebpf/src/main.rs +++ b/lnvps_fw/lnvps_ebpf/src/main.rs @@ -23,8 +23,8 @@ use network_types::udp::UdpHdr; mod maps; use maps::{ - OPEN_PORTS_V4, OPEN_PORTS_V6, SYN_PROXY_JUMP, cidr_blocked_v4, cidr_blocked_v6, cookie_secrets, - count_src_v4, count_src_v6, counters_v4, counters_v6, dest_mode_v4, dest_mode_v6, + OPEN_PORTS_V4, OPEN_PORTS_V6, SYN_PROXY_JUMP, cookie_secrets, counters_v4, counters_v6, + dest_mode_v4, dest_mode_v6, src_gate_v4, src_gate_v6, learn_port_v4, learn_port_v6, manual_blocked_v4, manual_blocked_v6, mark_verified_v4, mark_verified_v6, port_is_open_v4, port_is_open_v6, protected_v4, protected_v6, scoped, src_verified_v4, src_verified_v6, tx_counters_v4, tx_counters_v6, @@ -257,8 +257,10 @@ fn mitigate_v4( allow_syn_proxy: bool, counters: Option<*mut lnvps_fw_common::DestCounters>, ) -> (u32, bool) { - count_src_v4(src); - if flags & DEST_MODE_SOURCE_BLOCK != 0 && cidr_blocked_v4(*src) { + // In-kernel per-source rate machine: counts this packet against the + // source's window and drops while the source is blocked (enforcement is + // gated on the dest's SOURCE_BLOCK escalation, counting is not). + if src_gate_v4(src, flags & DEST_MODE_SOURCE_BLOCK != 0) { return (XDP_DROP, false); } if allow_syn_proxy @@ -312,8 +314,7 @@ fn mitigate_v6( allow_syn_proxy: bool, counters: Option<*mut lnvps_fw_common::DestCounters>, ) -> (u32, bool) { - count_src_v6(src); - if flags & DEST_MODE_SOURCE_BLOCK != 0 && cidr_blocked_v6(*src) { + if src_gate_v6(src, flags & DEST_MODE_SOURCE_BLOCK != 0) { return (XDP_DROP, false); } if allow_syn_proxy diff --git a/lnvps_fw/lnvps_ebpf/src/maps.rs b/lnvps_fw/lnvps_ebpf/src/maps.rs index 02e7f13d..757ea80e 100644 --- a/lnvps_fw/lnvps_ebpf/src/maps.rs +++ b/lnvps_fw/lnvps_ebpf/src/maps.rs @@ -4,7 +4,7 @@ use aya_ebpf::maps::lpm_trie::Key; use aya_ebpf::maps::{Array, LpmTrie, LruHashMap, LruPerCpuHashMap, ProgramArray}; use lnvps_fw_common::{ COOKIE_SECRET_CURRENT, COOKIE_SECRET_PREVIOUS, DEST_MODE_NORMAL, DestCounters, DestState, - LastSeen, PortKeyV4, PortKeyV6, + LastSeen, PortKeyV4, PortKeyV6, SrcRateConfig, SrcState, }; /// Max number of destination IPs to track (per address family). Sized to a @@ -48,17 +48,28 @@ pub static V4_TX_COUNTERS: LruPerCpuHashMap<[u8; 4], DestCounters> = pub static V6_TX_COUNTERS: LruPerCpuHashMap<[u8; 16], DestCounters> = LruPerCpuHashMap::with_max_entries(MAX_DST_IPS, 0); -/// Per-source packet counters (IPv4), incremented only while the destination -/// is mitigating. Sampled by userspace to compute per-source rates and drive -/// CIDR escalation. LRU + per-CPU: bounded memory, summed across CPUs. +/// Per-source fixed-window rate state (IPv4), owned by the XDP datapath: the +/// rate calculation AND the blocking decision happen in-kernel, in-line, for +/// packets already in hand — userspace never polls this for detection, it only +/// reads it (batched) for the `/sources` display. LRU: bounded memory; a +/// spoofed high-cardinality flood churns entries without ever tripping the +/// per-source limit (the port-filter layer is the defense there, as before). +/// Shared (not per-CPU): counting races undercount slightly, which is +/// acceptable — the decision uses the same state the packets update. #[map] -pub static V4_SRC_COUNTERS: LruPerCpuHashMap<[u8; 4], u64> = - LruPerCpuHashMap::with_max_entries(MAX_SRC_IPS, 0); +pub static V4_SRC_STATE: LruHashMap<[u8; 4], SrcState> = + LruHashMap::with_max_entries(MAX_SRC_IPS, 0); -/// Per-source packet counters (IPv6). +/// Per-source rate state (IPv6). #[map] -pub static V6_SRC_COUNTERS: LruPerCpuHashMap<[u8; 16], u64> = - LruPerCpuHashMap::with_max_entries(MAX_SRC_IPS, 0); +pub static V6_SRC_STATE: LruHashMap<[u8; 16], SrcState> = + LruHashMap::with_max_entries(MAX_SRC_IPS, 0); + +/// Per-source rate-machine config (single entry), written by userspace at +/// startup and on live `PUT /limits` edits. `max_per_window == 0` disables +/// per-source blocking. +#[map] +pub static SRC_RATE_CFG: Array = Array::with_max_entries(1, 0); /// Mitigation state per destination (IPv4), an LPM trie written by userspace. /// Using a trie lets userspace mitigate a single IP (a /32 entry) or a whole @@ -81,16 +92,6 @@ pub static OPEN_PORTS_V4: LruHashMap = pub static OPEN_PORTS_V6: LruHashMap = LruHashMap::with_max_entries(MAX_OPEN_PORTS, 0); -/// Blocked source CIDRs (IPv4), an LPM trie of network-order address bytes. -/// Written by userspace escalation; any source matching a prefix is dropped. -/// Holds both individual /32 offenders and aggregated wider prefixes. -#[map] -pub static V4_CIDR_SRC: LpmTrie<[u8; 4], u8> = LpmTrie::with_max_entries(MAX_CIDR_BLOCKS, 0); - -/// Blocked source CIDRs (IPv6). -#[map] -pub static V6_CIDR_SRC: LpmTrie<[u8; 16], u8> = LpmTrie::with_max_entries(MAX_CIDR_BLOCKS, 0); - /// Protected destination prefixes (IPv4). When scoping is enabled, only traffic /// to a covered destination is counted/mitigated; everything else is passed /// untouched (so a forwarding router never touches transit traffic). @@ -101,9 +102,10 @@ pub static PROTECTED_V4: LpmTrie<[u8; 4], u8> = LpmTrie::with_max_entries(MAX_CI #[map] pub static PROTECTED_V6: LpmTrie<[u8; 16], u8> = LpmTrie::with_max_entries(MAX_CIDR_BLOCKS, 0); -/// Operator-pushed manual source-CIDR blocks. Unlike the auto `V*_CIDR_SRC` -/// blocks (which only drop when the destination is escalated to SOURCE_BLOCK), -/// these drop unconditionally for any traffic to a protected destination. +/// Operator-pushed manual source-CIDR blocks. Unlike the automatic per-source +/// rate gate (whose drops engage only when the destination is escalated to +/// SOURCE_BLOCK), these drop unconditionally for any traffic to a protected +/// destination. #[map] pub static MANUAL_BLOCK_V4: LpmTrie<[u8; 4], u8> = LpmTrie::with_max_entries(MAX_CIDR_BLOCKS, 0); @@ -163,24 +165,64 @@ counters_for!(counters_v6, [u8; 16], V6_DEST_COUNTERS); counters_for!(tx_counters_v4, [u8; 4], V4_TX_COUNTERS); counters_for!(tx_counters_v6, [u8; 16], V6_TX_COUNTERS); -/// Generate a per-source packet-count incrementer for one address family -/// (called under mitigation). Pure counting — no policy decision. -macro_rules! count_src_for { +/// Generate the per-source fixed-window rate gate for one address family. +/// Called for every packet to a mitigating destination. Counts the packet +/// against the source's current window and returns `true` (drop) when +/// `enforce` is set and the source is blocked. +/// +/// The machine, entirely in-kernel: +/// - window rolled (`now - window_start >= window_ns`): reset the count; +/// - count the packet; crossing `max_per_window` sets/extends +/// `blocked_until = now + cooldown` (a still-flooding source re-extends its +/// block every window, so the block naturally outlives the flood by exactly +/// one cooldown — that is the hysteresis); +/// - blocked sources keep being *counted* (the rate is measured pre-drop) so +/// expiry re-evaluates against live behaviour, not silence. +/// +/// Counting is deliberately non-atomic (same as the counters this replaces): +/// cross-CPU races undercount a few packets, which only ever errs toward NOT +/// blocking. +macro_rules! src_gate_for { ($name:ident, $key:ty, $map:ident) => { #[inline(always)] - pub fn $name(src: &$key) { - if let Some(c) = $map.get_ptr_mut(src) { - unsafe { *c += 1 }; - } else { - let one: u64 = 1; - let _ = $map.insert(src, &one, 0); + pub fn $name(src: &$key, enforce: bool) -> bool { + let cfg = match SRC_RATE_CFG.get(0) { + Some(c) => c, + None => return false, + }; + if cfg.max_per_window == 0 || cfg.window_ns == 0 { + return false; + } + let now = unsafe { bpf_ktime_get_ns() }; + match $map.get_ptr_mut(src) { + Some(st) => { + let st = unsafe { &mut *st }; + if now.wrapping_sub(st.window_start_ns) >= cfg.window_ns { + st.window_start_ns = now; + st.count = 0; + } + st.count += 1; + if st.count > cfg.max_per_window { + st.blocked_until_ns = now + cfg.cooldown_ns; + } + enforce && now < st.blocked_until_ns + } + None => { + let st = SrcState { + window_start_ns: now, + count: 1, + blocked_until_ns: 0, + }; + let _ = $map.insert(src, &st, 0); + false + } } } }; } -count_src_for!(count_src_v4, [u8; 4], V4_SRC_COUNTERS); -count_src_for!(count_src_v6, [u8; 16], V6_SRC_COUNTERS); +src_gate_for!(src_gate_v4, [u8; 4], V4_SRC_STATE); +src_gate_for!(src_gate_v6, [u8; 16], V6_SRC_STATE); /// Generate a learn-open-port function for one address family. Called from the /// TC egress program with the local (source) address/port of an outbound @@ -236,8 +278,6 @@ macro_rules! cidr_block_check { }; } -cidr_block_check!(cidr_blocked_v4, [u8; 4], 32, V4_CIDR_SRC); -cidr_block_check!(cidr_blocked_v6, [u8; 16], 128, V6_CIDR_SRC); cidr_block_check!(protected_v4, [u8; 4], 32, PROTECTED_V4); cidr_block_check!(protected_v6, [u8; 16], 128, PROTECTED_V6); cidr_block_check!(manual_blocked_v4, [u8; 4], 32, MANUAL_BLOCK_V4); diff --git a/lnvps_fw/lnvps_fw_common/src/lib.rs b/lnvps_fw/lnvps_fw_common/src/lib.rs index f5daf910..c6279090 100644 --- a/lnvps_fw/lnvps_fw_common/src/lib.rs +++ b/lnvps_fw/lnvps_fw_common/src/lib.rs @@ -64,6 +64,37 @@ pub struct DestCounters { pub dropped: u64, } +/// Per-source fixed-window rate state, owned entirely by the XDP datapath. +/// The kernel computes the rate and the blocking decision in-line for packets +/// it is already touching; userspace only reads this for display. +#[repr(C)] +#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)] +pub struct SrcState { + /// Monotonic (`bpf_ktime_get_ns`) anchor of the current counting window. + pub window_start_ns: u64, + /// Packets counted in the current window (atomic add across CPUs). + pub count: u64, + /// Drop packets from this source until this timestamp (0 = not blocked). + /// Re-trips (extends) if the source is still over-rate when it expires. + pub blocked_until_ns: u64, +} + +/// Per-source rate-machine configuration, written by userspace (startup + +/// live `PUT /limits`), read by XDP per packet under mitigation. One entry in +/// an `Array` map. +#[repr(C)] +#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)] +pub struct SrcRateConfig { + /// A source exceeding this many packets within one window is blocked. + /// Precomputed by userspace as `rate_pps × window_secs` so the datapath + /// never divides. 0 disables per-source blocking entirely. + pub max_per_window: u64, + /// Fixed counting-window length. + pub window_ns: u64, + /// How long a tripped source stays blocked before re-evaluation. + pub cooldown_ns: u64, +} + /// Mitigation state for a destination IP. Written by userspace (detection /// state machine), read by the XDP ingress program. #[repr(C)] @@ -253,6 +284,8 @@ mod user { unsafe impl aya::Pod for DestCounters {} unsafe impl aya::Pod for DestState {} + unsafe impl aya::Pod for SrcState {} + unsafe impl aya::Pod for SrcRateConfig {} unsafe impl aya::Pod for LastSeen {} unsafe impl aya::Pod for PortKeyV4 {} unsafe impl aya::Pod for PortKeyV6 {} diff --git a/lnvps_fw/lnvps_fw_service/config.example.yaml b/lnvps_fw/lnvps_fw_service/config.example.yaml index 1bfd2258..78953b01 100644 --- a/lnvps_fw/lnvps_fw_service/config.example.yaml +++ b/lnvps_fw/lnvps_fw_service/config.example.yaml @@ -79,52 +79,28 @@ thresholds: # How often the detection loop samples the per-destination counters. sample-interval-ms: 500 -# Per-source detection + CIDR aggregation while a destination is mitigating. -# NOTE: the eBPF datapath only COUNTS per source (bounded LRU) and enforces the -# CIDR block trie; all rate math and block decisions are made here in userspace. -# This is a secondary defence aimed at real (unspoofed) botnets; spoofed -# carpet-bomb floods are handled by per-destination mitigation + the -# drop-to-non-open-ports policy, not by source blocking. +# Per-source rate limiting while a destination is mitigating. +# The rate machine lives IN the XDP datapath: the kernel counts each source +# over a fixed 1s window and blocks over-rate sources in-line (drops engage +# once the destination escalates to SOURCE_BLOCK). Userspace only writes this +# config into the datapath and reads the state for display — it is not in the +# decision path. This is a defence against real (unspoofed) botnets; spoofed +# floods never trip a per-source limit and are handled by the port-filter / +# SYN-proxy layers. escalation: - # A source is an offender once its per-window rate reaches this many pps - # toward a mitigated destination. Keep this well above what shared + # A source is blocked once it sends more than this many packets within one + # 1s window toward a mitigated destination. Keep this well above what shared # infrastructure (CDN/reverse-proxy edges, CGNAT, corporate NAT) legitimately # sends to one origin — a busy CDN edge easily sustains hundreds–thousands of # pps. Also runtime-editable via PUT /api/v1/limits (src_rate_pps). src-rate-pps: 10000 - # Aggregation fan-out: this many child prefixes under a common parent - # collapse into one wider block (/32->/24->/16->/8), keeping the LPM trie - # bounded under distributed floods. - agg-fanout: 4 - # Widest block aggregation may ever produce (smallest prefix length). The - # default /24 (v4) / /48 (v6) stops a scatter of offenders — e.g. legitimate - # CDN edge IPs — from ever collapsing into an allocation-crossing /16 or /8 - # and blackholing huge swaths of the internet. Only widen these if you fully - # understand the collateral risk. - agg-max-prefix-v4: 24 - agg-max-prefix-v6: 48 - # Per-source block hysteresis: a source is DROPPING (in the block trie) only - # while its rate is high. It is released once its rate falls below - # src-exit-pct% of src-rate-pps for src-cooldown-secs — so a source that had - # one burst and returned to legitimate low-rate traffic is unblocked quickly, - # not held for the full TTL. - src-exit-pct: 50 + # How long a tripped source stays blocked before re-evaluation. A source + # still over-rate re-extends its block every window, so the block outlives + # the flood by exactly one cooldown. src-cooldown-secs: 10 - # Block offending sources as individual /32s (/128s) up to this many entries; - # only merge them into wider prefixes when the trie would exceed this budget - # ("merge only when running out of space"). Keeps blocks precise in the common - # case and avoids catching innocent neighbours in an aggregated prefix. - max-source-blocks: 50000 - # Safety upper-bound: a block is force-lifted this many seconds after it stops - # being refreshed (the per-source state machine above is the primary release). - block-ttl-secs: 300 - # Escalate a mitigating dest/prefix to source blocking (level 4) only if this - # many packets/second are still getting through after the port-filter layer. + # Escalate a mitigating dest/prefix to source blocking only if this many + # packets/second are still getting through after the port-filter layer. escalate-pass-pps: 50000 - # Spoof gate: if more than this many distinct offenders appear in a window, - # treat the flood as spoofed and skip source blocking entirely (rely on the - # port filter). Source blocking only makes sense for bounded, real botnets. - max-real-sources: 10000 # Enable the SYN-proxy (SYN-cookie handshake validation) on a mitigating # dest/prefix once its SYN rate reaches this many SYNs/second. syn-proxy-syn-pps: 5000 diff --git a/lnvps_fw/lnvps_fw_service/dashboard/dist/index.html b/lnvps_fw/lnvps_fw_service/dashboard/dist/index.html index c329d21a..d662f2d7 100644 --- a/lnvps_fw/lnvps_fw_service/dashboard/dist/index.html +++ b/lnvps_fw/lnvps_fw_service/dashboard/dist/index.html @@ -4,7 +4,7 @@ lnvps_fw dashboard - + diff --git a/lnvps_fw/lnvps_fw_service/dashboard/src/api.ts b/lnvps_fw/lnvps_fw_service/dashboard/src/api.ts index 7f210a91..e24c3491 100644 --- a/lnvps_fw/lnvps_fw_service/dashboard/src/api.ts +++ b/lnvps_fw/lnvps_fw_service/dashboard/src/api.ts @@ -26,7 +26,7 @@ export interface Mitigation { } export interface SourceBlock { cidr: string; age_secs: number; pps: number; manual: boolean; cooling: boolean } /** A row in the unified source list: tracked sources (any state) + manual blocks. */ -export interface TrackedSource { ip: string; pps: number; state: "normal" | "dropping" | "cooling"; manual: boolean; age_secs: number } +export interface TrackedSource { ip: string; pps: number; state: "normal" | "dropping"; manual: boolean; age_secs: number } export interface LearnedPort { ip: string; port: number; proto: string; age_secs: number } export interface FwEvent { seq: number; kind: string; cidr: string; flags: number; ts_unix: number; @@ -35,7 +35,7 @@ export interface FwEvent { export interface Limits { pps: number; syn_pps: number; bps: number; net_pps: number; net_syn_pps: number; net_bps: number; exit_pct: number; cooldown_secs: number; - src_rate_pps: number; src_exit_pct: number; src_cooldown_secs: number; + src_rate_pps: number; src_cooldown_secs: number; } export interface UpgradeStatus { current: string; latest: string | null; available: boolean; diff --git a/lnvps_fw/lnvps_fw_service/dashboard/src/cards.tsx b/lnvps_fw/lnvps_fw_service/dashboard/src/cards.tsx index 5663b18a..3ba6784c 100644 --- a/lnvps_fw/lnvps_fw_service/dashboard/src/cards.tsx +++ b/lnvps_fw/lnvps_fw_service/dashboard/src/cards.tsx @@ -84,7 +84,7 @@ export function LimitsCard({ token, nics }: { token: string; nics?: InterfaceInf {rateFld("pps", "IP pps")}{rateFld("syn_pps", "IP syn/s")}{bpsFld("bps", "IP bit/s")} {rateFld("net_pps", "prefix pps")}{rateFld("net_syn_pps", "prefix syn/s")}{bpsFld("net_bps", "prefix bit/s")} {fld("exit_pct", "exit %")}{fld("cooldown_secs", "cooldown s")} - {rateFld("src_rate_pps", "src block pps")}{fld("src_exit_pct", "src exit %")}{fld("src_cooldown_secs", "src cooldown s")} + {rateFld("src_rate_pps", "src block pps")}{fld("src_cooldown_secs", "src cooldown s")}
@@ -152,7 +152,7 @@ export function MitigationsCard({ token, mitigations, onChange }: { ); } -// --- Sources: unified list of rate-tracked sources (normal/dropping/cooling) +// --- Sources: unified list of rate-tracked sources (normal/dropping) // + manual blocks. Server-paginated + filtered. Auto "blocks" are just the // dropping/cooling rows here — there is no separate block list. --- export function SourcesCard({ token }: { token: string }) { @@ -182,7 +182,6 @@ export function SourcesCard({ token }: { token: string }) { const bin = (c: string) => ; const stateCell = (s: TrackedSource) => s.manual ? pinned - : s.state === "cooling" ? cooling : s.state === "dropping" ? dropping : normal; const pages = Math.max(1, Math.ceil(data.total / PAGE)); diff --git a/lnvps_fw/lnvps_fw_service/examples/serve_api.rs b/lnvps_fw/lnvps_fw_service/examples/serve_api.rs index 31f898ac..09767ec8 100644 --- a/lnvps_fw/lnvps_fw_service/examples/serve_api.rs +++ b/lnvps_fw/lnvps_fw_service/examples/serve_api.rs @@ -31,7 +31,6 @@ const LIM: Limits = Limits { exit_pct: 50, cooldown_secs: 30, src_rate_pps: 10_000, - src_exit_pct: 50, src_cooldown_secs: 10, }; const SYN_PROXY_PPS: u64 = 5_000; diff --git a/lnvps_fw/lnvps_fw_service/src/api.rs b/lnvps_fw/lnvps_fw_service/src/api.rs index 58292611..e585ae72 100644 --- a/lnvps_fw/lnvps_fw_service/src/api.rs +++ b/lnvps_fw/lnvps_fw_service/src/api.rs @@ -238,15 +238,13 @@ pub struct Limits { pub exit_pct: u64, pub cooldown_secs: u64, /// Per-source auto-block threshold: once a destination is mitigating, any - /// single source at/over this pps is blocked (`escalation.src_rate_pps`). - /// This is a much lower bar than the per-destination `pps` above — keep - /// them side by side so neither hides. + /// single source at/over this pps (exact, over the kernel rate machine's + /// 1s window) is blocked. This is a much lower bar than the + /// per-destination `pps` above — kept side by side so neither hides. #[serde(default = "default_limit_src_rate_pps")] pub src_rate_pps: u64, - /// Source exit hysteresis (% of `src_rate_pps`). - #[serde(default = "default_limit_src_exit_pct")] - pub src_exit_pct: u64, - /// Sustained seconds below the source exit threshold before unblocking. + /// How long a tripped source stays blocked before re-evaluation (the + /// kernel re-extends the block each window the source is still over-rate). #[serde(default = "default_limit_src_cooldown_secs")] pub src_cooldown_secs: u64, } @@ -254,9 +252,6 @@ pub struct Limits { fn default_limit_src_rate_pps() -> u64 { 10_000 } -fn default_limit_src_exit_pct() -> u64 { - 50 -} fn default_limit_src_cooldown_secs() -> u64 { 10 } @@ -1003,9 +998,6 @@ async fn put_limits(State(state): State>, Json(l): Json if l.exit_pct == 0 || l.exit_pct >= 100 { return (StatusCode::BAD_REQUEST, "exit_pct must be 1..99").into_response(); } - if l.src_exit_pct == 0 || l.src_exit_pct >= 100 { - return (StatusCode::BAD_REQUEST, "src_exit_pct must be 1..99").into_response(); - } *state.limits.write().unwrap() = l; state.limits_version.fetch_add(1, Ordering::Relaxed); StatusCode::NO_CONTENT.into_response() diff --git a/lnvps_fw/lnvps_fw_service/src/batch.rs b/lnvps_fw/lnvps_fw_service/src/batch.rs new file mode 100644 index 00000000..aa1aa129 --- /dev/null +++ b/lnvps_fw/lnvps_fw_service/src/batch.rs @@ -0,0 +1,324 @@ +//! Batched BPF map reads via the raw `BPF_MAP_LOOKUP_BATCH` syscall. +//! +//! aya 0.14 only exposes per-entry iteration: every entry costs a +//! `BPF_MAP_GET_NEXT_KEY` + `BPF_MAP_LOOKUP_ELEM` syscall pair. The control +//! loop reads the source/dest counter maps every sample tick, so on a large +//! map (the source counters hold up to 256k entries during a flood) that is +//! ~1M syscalls/second — a pinned core. `BPF_MAP_LOOKUP_BATCH` (kernel 5.6+) +//! returns thousands of entries per syscall instead. +//! +//! Callers must fall back to per-entry iteration when [`supported`] turns +//! false: the first syscall failure with `EINVAL`/`ENOTSUPP` (old kernel, or +//! a map type without batch support) latches the flag off process-wide. + +use std::io; +use std::mem::size_of; +use std::os::fd::{AsRawFd, BorrowedFd}; +use std::sync::atomic::{AtomicBool, Ordering}; + +use aya::Pod; + +/// `enum bpf_cmd` value for `BPF_MAP_LOOKUP_BATCH`. +const BPF_MAP_LOOKUP_BATCH: libc::c_long = 24; + +/// Entries requested per syscall. Large enough that syscall overhead is noise +/// (a full 256k-entry map is ~64 calls), small enough that the key/value +/// buffers stay modest (16-byte keys × 4096 = 64 KiB). +const CHUNK: usize = 4096; + +/// Kernel-visible tail of `union bpf_attr` used by the `BPF_MAP_*_BATCH` +/// commands (uapi `linux/bpf.h`). Field order and alignment must match. +#[repr(C)] +#[derive(Clone, Copy, Default)] +struct BatchAttr { + in_batch: u64, + out_batch: u64, + keys: u64, + values: u64, + count: u32, + map_fd: u32, + elem_flags: u64, + flags: u64, +} + +/// Latched process-wide support flag (assume supported until proven wrong). +static SUPPORTED: AtomicBool = AtomicBool::new(true); + +/// Whether batch lookups are (still believed to be) supported. Callers should +/// take their per-entry fallback path when this is false. +pub fn supported() -> bool { + SUPPORTED.load(Ordering::Relaxed) +} + +/// Does this error mean "the kernel or map type cannot do batch lookups" +/// (permanent — latch off and fall back) rather than a transient failure? +fn is_unsupported(e: &io::Error) -> bool { + // ENOTSUPP (524) has no libc constant; EINVAL = kernel predates the + // command or rejects it for this map type; EPERM under a locked-down bpf(). + matches!( + e.raw_os_error(), + Some(libc::EINVAL) | Some(libc::EPERM) | Some(524) + ) +} + +/// One raw `BPF_MAP_LOOKUP_BATCH` call. Returns `Ok(done)` where `done` means +/// the map is exhausted (`ENOENT` — the final partial chunk, if any, is still +/// delivered). `attr.count` is updated by the kernel to the entries returned. +/// +/// # Safety +/// `attr` must point key/value buffers with room for `attr.count` entries of +/// the map's key size / value stride respectively, and `in_batch`/`out_batch` +/// at valid token storage. +unsafe fn lookup_batch_once(attr: &mut BatchAttr) -> io::Result { + let ret = unsafe { + libc::syscall( + libc::SYS_bpf, + BPF_MAP_LOOKUP_BATCH, + attr as *mut BatchAttr, + size_of::(), + ) + }; + if ret == 0 { + return Ok(false); + } + let err = io::Error::last_os_error(); + if err.raw_os_error() == Some(libc::ENOENT) { + return Ok(true); // exhausted; attr.count holds the final chunk size + } + Err(err) +} + +/// Scan an entire hash-family map with batched lookups into raw byte buffers. +/// `stride` is the byte size of one entry's value block (`value_size` for +/// plain maps, `round8(value_size) × possible_cpus` for per-CPU maps). +/// Returns `(keys, values, n)` holding `n` densely-packed entries. +fn scan_raw( + fd: BorrowedFd<'_>, + key_size: usize, + stride: usize, +) -> io::Result<(Vec, Vec, usize)> { + let mut keys: Vec = Vec::new(); + let mut vals: Vec = Vec::new(); + let mut total = 0usize; + // Opaque per-map resume token (a bucket index for hash maps; u64 storage + // is large enough for every map family). + let mut token_in = 0u64; + let mut token_out = 0u64; + let mut first = true; + let mut chunk = CHUNK; + loop { + keys.resize((total + chunk) * key_size, 0); + vals.resize((total + chunk) * stride, 0); + let mut attr = BatchAttr { + in_batch: if first { 0 } else { &raw mut token_in as u64 }, + out_batch: &raw mut token_out as u64, + keys: keys[total * key_size..].as_mut_ptr() as u64, + values: vals[total * stride..].as_mut_ptr() as u64, + count: chunk as u32, + map_fd: fd.as_raw_fd() as u32, + elem_flags: 0, + flags: 0, + }; + // SAFETY: buffers sized for `chunk` entries above; tokens are valid. + let done = match unsafe { lookup_batch_once(&mut attr) } { + Ok(done) => done, + // A bucket larger than the whole chunk (pathological collisions): + // grow and retry the same position. + Err(e) if e.raw_os_error() == Some(libc::ENOSPC) => { + chunk *= 2; + continue; + } + Err(e) => return Err(e), + }; + total += attr.count as usize; + if done { + break; + } + token_in = token_out; + first = false; + } + keys.truncate(total * key_size); + vals.truncate(total * stride); + Ok((keys, vals, total)) +} + +/// Round a per-CPU value size up to the kernel's 8-byte per-CPU slot stride. +fn round8(n: usize) -> usize { + n.div_ceil(8) * 8 +} + +/// Batched read of a **per-CPU** map, folding each entry's per-CPU values into +/// one `T` with `fold` (e.g. summing counters across CPUs). +pub fn read_percpu_folded( + fd: BorrowedFd<'_>, + ncpus: usize, + fold: impl Fn(&mut T, &V), +) -> io::Result> +where + K: Pod, + V: Pod, + T: Default, +{ + let key_size = size_of::(); + let slot = round8(size_of::()); + let stride = slot * ncpus; + let (keys, vals, n) = scan_raw(fd, key_size, stride)?; + let mut out = Vec::with_capacity(n); + for i in 0..n { + // SAFETY: buffers hold `n` densely-packed entries; K/V are Pod, and + // every per-CPU slot lies within the entry's stride block. + let k = unsafe { *(keys[i * key_size..].as_ptr() as *const K) }; + let mut acc = T::default(); + for cpu in 0..ncpus { + let v = unsafe { *(vals[i * stride + cpu * slot..].as_ptr() as *const V) }; + fold(&mut acc, &v); + } + out.push((k, acc)); + } + Ok(out) +} + +/// Batched read of a **plain** (non-per-CPU) map into `(key, value)` pairs. +pub fn read_plain(fd: BorrowedFd<'_>) -> io::Result> +where + K: Pod, + V: Pod, +{ + let key_size = size_of::(); + let stride = size_of::(); + let (keys, vals, n) = scan_raw(fd, key_size, stride)?; + let mut out = Vec::with_capacity(n); + for i in 0..n { + // SAFETY: buffers hold `n` densely-packed entries and K/V are Pod. + let k = unsafe { *(keys[i * key_size..].as_ptr() as *const K) }; + let v = unsafe { *(vals[i * stride..].as_ptr() as *const V) }; + out.push((k, v)); + } + Ok(out) +} + +/// Record a batch failure: permanent unsupport latches the flag off (callers +/// fall back to per-entry iteration for the rest of the process lifetime). +/// Returns true if the error was the permanent kind. +pub fn note_failure(e: &io::Error) -> bool { + if is_unsupported(e) { + SUPPORTED.store(false, Ordering::Relaxed); + return true; + } + false +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round8_covers_slot_alignment() { + assert_eq!(round8(1), 8); + assert_eq!(round8(8), 8); + assert_eq!(round8(9), 16); + assert_eq!(round8(56), 56); // DestCounters: 7 × u64 + } + + #[test] + fn unsupported_errors_latch_off() { + assert!(is_unsupported(&io::Error::from_raw_os_error(libc::EINVAL))); + assert!(is_unsupported(&io::Error::from_raw_os_error(524))); // ENOTSUPP + assert!(!is_unsupported(&io::Error::from_raw_os_error(libc::ENOENT))); + assert!(!is_unsupported(&io::Error::from_raw_os_error(libc::EFAULT))); + } + + #[test] + fn batch_attr_layout_matches_uapi() { + // Offsets from uapi linux/bpf.h `struct bpf_attr { ... } batch;` + assert_eq!(std::mem::offset_of!(BatchAttr, in_batch), 0); + assert_eq!(std::mem::offset_of!(BatchAttr, out_batch), 8); + assert_eq!(std::mem::offset_of!(BatchAttr, keys), 16); + assert_eq!(std::mem::offset_of!(BatchAttr, values), 24); + assert_eq!(std::mem::offset_of!(BatchAttr, count), 32); + assert_eq!(std::mem::offset_of!(BatchAttr, map_fd), 36); + assert_eq!(std::mem::offset_of!(BatchAttr, elem_flags), 40); + assert_eq!(std::mem::offset_of!(BatchAttr, flags), 48); + assert_eq!(size_of::(), 56); + } + + /// End-to-end against a real kernel map (needs CAP_BPF/root): create an + /// LRU per-CPU hash, fill it via aya, batch-read it back and compare with + /// aya's per-entry iteration. + #[test] + #[ignore = "requires root/CAP_BPF"] + fn batch_read_matches_iteration_on_real_map() { + use std::os::fd::FromRawFd; + // Create a BPF_MAP_TYPE_LRU_PERCPU_HASH (9) directly via bpf(2). + #[repr(C)] + #[derive(Default)] + struct CreateAttr { + map_type: u32, + key_size: u32, + value_size: u32, + max_entries: u32, + map_flags: u32, + } + let attr = CreateAttr { + map_type: 10, // BPF_MAP_TYPE_LRU_PERCPU_HASH + key_size: 4, + value_size: 8, + max_entries: 1024, + ..Default::default() + }; + let fd = unsafe { + libc::syscall( + libc::SYS_bpf, + 0i64, // BPF_MAP_CREATE + &attr as *const CreateAttr, + size_of::(), + ) + }; + assert!(fd >= 0, "map create failed: {}", io::Error::last_os_error()); + let owned = unsafe { std::os::fd::OwnedFd::from_raw_fd(fd as i32) }; + let ncpus = aya::util::nr_cpus().expect("nr_cpus"); + + // Insert 100 keys, value = key on every CPU slot. + #[repr(C)] + struct UpdateAttr { + map_fd: u32, + _pad: u32, + key: u64, + value: u64, + flags: u64, + } + let slot = round8(8); + for i in 0u32..100 { + let vals: Vec = (0..ncpus) + .flat_map(|_| u64::from(i).to_ne_bytes()) + .collect(); + assert_eq!(vals.len(), slot * ncpus); + let ua = UpdateAttr { + map_fd: fd as u32, + _pad: 0, + key: &raw const i as u64, + value: vals.as_ptr() as u64, + flags: 0, + }; + let r = unsafe { + libc::syscall( + libc::SYS_bpf, + 1i64, // BPF_MAP_UPDATE_ELEM + &ua as *const UpdateAttr, + size_of::(), + ) + }; + assert_eq!(r, 0, "update failed: {}", io::Error::last_os_error()); + } + + use std::os::fd::AsFd; + let got: Vec<([u8; 4], u64)> = + read_percpu_folded(owned.as_fd(), ncpus, |acc: &mut u64, v: &u64| *acc += *v) + .expect("batch read"); + assert_eq!(got.len(), 100); + for (k, sum) in got { + let key = u32::from_ne_bytes(k); + assert_eq!(sum, u64::from(key) * ncpus as u64); + } + } +} diff --git a/lnvps_fw/lnvps_fw_service/src/cidr.rs b/lnvps_fw/lnvps_fw_service/src/cidr.rs index b20a1a3c..26a6937b 100644 --- a/lnvps_fw/lnvps_fw_service/src/cidr.rs +++ b/lnvps_fw/lnvps_fw_service/src/cidr.rs @@ -14,9 +14,6 @@ //! //! Keeping this free of BPF handles makes it fully unit-testable. -use std::collections::HashMap; -use std::hash::Hash; - /// A blocked IPv4 CIDR: prefix length + network address (bits below the prefix /// zeroed, bytes exactly as on the wire). #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -32,14 +29,6 @@ pub struct CidrV6 { pub network: [u8; 16], } -/// IPv4 aggregation levels, widest cap first is applied last: offenders start -/// as /32 and collapse toward /8 — but only levels at least as narrow as the -/// configured `max_prefix` cap are ever applied (see [`aggregate_v4`]), so the -/// default cap keeps blocks from ever crossing large allocation boundaries. -const V4_LEVELS: [u32; 3] = [24, 16, 8]; -/// IPv6 aggregation levels: offenders start as /128 and collapse toward /32. -const V6_LEVELS: [u32; 3] = [64, 48, 32]; - /// Safe default widest IPv4 aggregation prefix (smallest prefix length). A /24 /// is the smallest globally-routable unit and is typically single-org, so /// aggregation never blackholes across organisation / RIR-allocation @@ -77,382 +66,24 @@ fn mask_bytes(addr: [u8; N], prefix: u32) -> [u8; N] { out } -/// Compute per-source packets/second from the previous and current cumulative -/// counter snapshots. Sources whose counters reset (LRU eviction) yield 0. -pub fn per_source_pps( - prev: &HashMap, - cur: &[(K, u64)], - elapsed_ns: u64, -) -> Vec<(K, u64)> { - if elapsed_ns == 0 { - return Vec::new(); - } - cur.iter() - .map(|(k, c)| { - let p = prev.get(k).copied().unwrap_or(0); - let delta = c.saturating_sub(p); - let pps = ((delta as u128 * 1_000_000_000u128) / elapsed_ns as u128) as u64; - (*k, pps) - }) - .collect() -} - -/// Select the sources whose per-second rate is at or above `min_pps`. -pub fn offenders(rates: &[(K, u64)], min_pps: u64) -> Vec { - rates - .iter() - .filter(|(_, pps)| *pps >= min_pps) - .map(|(k, _)| *k) - .collect() -} - -/// Aggregate offending IPv4 sources into the fewest CIDR blocks. A parent -/// prefix replaces its children once at least `fanout` distinct children fall -/// under it; otherwise the children are blocked individually. `fanout` of 0 or -/// 1 is treated as 2 (a single child never widens). -/// -/// `max_prefix` caps how wide a block may become (the smallest prefix length, -/// e.g. 24 = never wider than a /24). Only ladder levels `>= max_prefix` are -/// applied, so the block set can never expand past the cap and blackhole a -/// whole allocation. `max_prefix` is clamped to `1..=32`. -pub fn aggregate_v4(offenders: &[[u8; 4]], fanout: usize, max_prefix: u32) -> Vec { - let fanout = fanout.max(2); - let max_prefix = max_prefix.clamp(1, 32); - let mut blocks: Vec = dedup(offenders.iter().map(|a| CidrV4 { - prefix_len: 32, - network: *a, - })); - for &target in &V4_LEVELS { - if target < max_prefix { - continue; // would exceed the configured widest-block cap - } - blocks = collapse_v4(blocks, target, fanout); - } - blocks.sort_by(|a, b| (a.prefix_len, a.network).cmp(&(b.prefix_len, b.network))); - blocks -} - -/// IPv6 equivalent of [`aggregate_v4`]. `max_prefix` is clamped to `1..=128`. -pub fn aggregate_v6(offenders: &[[u8; 16]], fanout: usize, max_prefix: u32) -> Vec { - let fanout = fanout.max(2); - let max_prefix = max_prefix.clamp(1, 128); - let mut blocks: Vec = dedup(offenders.iter().map(|a| CidrV6 { - prefix_len: 128, - network: *a, - })); - for &target in &V6_LEVELS { - if target < max_prefix { - continue; - } - blocks = collapse_v6(blocks, target, fanout); - } - blocks.sort_by(|a, b| (a.prefix_len, a.network).cmp(&(b.prefix_len, b.network))); - blocks -} - -/// Plan the IPv4 source-block set from the currently-DROPPING source addresses. -/// -/// Individual `/32`s are preferred so each source is governed independently by -/// its own rate state machine. Aggregation is applied **only under trie space -/// pressure**: if the `/32` count exceeds `budget`, offenders are progressively -/// collapsed (densest `/24`s first, then wider) until the set fits — never past -/// the `max_prefix` safety cap. So in the normal case nothing is merged. -pub fn plan_blocks_v4( - dropping: &[[u8; 4]], - fanout: usize, - budget: usize, - max_prefix: u32, -) -> Vec { - let base: Vec = dedup(dropping.iter().map(|a| CidrV4 { - prefix_len: 32, - network: *a, - })); - if base.len() <= budget { - let mut b = base; - b.sort_by(|a, b| (a.prefix_len, a.network).cmp(&(b.prefix_len, b.network))); - return b; - } - // Under pressure: widen only as far as needed, never wider than max_prefix. - for &cap in &[24u32, 16, 8] { - if cap < max_prefix.clamp(1, 32) { - continue; - } - let blocks = aggregate_v4(dropping, fanout, cap); - if blocks.len() <= budget { - return blocks; - } - } - aggregate_v4(dropping, fanout, max_prefix) -} - -/// IPv6 counterpart of [`plan_blocks_v4`]. -pub fn plan_blocks_v6( - dropping: &[[u8; 16]], - fanout: usize, - budget: usize, - max_prefix: u32, -) -> Vec { - let base: Vec = dedup(dropping.iter().map(|a| CidrV6 { - prefix_len: 128, - network: *a, - })); - if base.len() <= budget { - let mut b = base; - b.sort_by(|a, b| (a.prefix_len, a.network).cmp(&(b.prefix_len, b.network))); - return b; - } - for &cap in &[64u32, 48, 32] { - if cap < max_prefix.clamp(1, 128) { - continue; - } - let blocks = aggregate_v6(dropping, fanout, cap); - if blocks.len() <= budget { - return blocks; - } - } - aggregate_v6(dropping, fanout, max_prefix) -} - -fn dedup(it: impl Iterator) -> Vec { - let mut seen: Vec = Vec::new(); - for c in it { - if !seen.contains(&c) { - seen.push(c); - } - } - seen -} - -fn collapse_v4(blocks: Vec, target: u32, fanout: usize) -> Vec { - let mut groups: HashMap<[u8; 4], Vec> = HashMap::new(); - let mut keep: Vec = Vec::new(); - for b in blocks { - if b.prefix_len > target { - groups - .entry(mask_v4(b.network, target)) - .or_default() - .push(b); - } else { - keep.push(b); - } - } - for (net, children) in groups { - if children.len() >= fanout { - keep.push(CidrV4 { - prefix_len: target, - network: net, - }); - } else { - keep.extend(children); - } - } - keep -} - -fn collapse_v6(blocks: Vec, target: u32, fanout: usize) -> Vec { - let mut groups: HashMap<[u8; 16], Vec> = HashMap::new(); - let mut keep: Vec = Vec::new(); - for b in blocks { - if b.prefix_len > target { - groups - .entry(mask_v6(b.network, target)) - .or_default() - .push(b); - } else { - keep.push(b); - } - } - for (net, children) in groups { - if children.len() >= fanout { - keep.push(CidrV6 { - prefix_len: target, - network: net, - }); - } else { - keep.extend(children); - } - } - keep -} - #[cfg(test)] mod tests { use super::*; #[test] - fn per_source_pps_and_offenders() { - let mut prev = HashMap::new(); - prev.insert([1, 1, 1, 1], 100u64); - let cur = [([1, 1, 1, 1], 700), ([2, 2, 2, 2], 50)]; - let rates = per_source_pps(&prev, &cur, 1_000_000_000); - // .1.1.1.1: +600 => 600pps; .2.2.2.2: new => 50pps - let off = offenders(&rates, 500); - assert_eq!(off, vec![[1, 1, 1, 1]]); - } - - #[test] - fn single_offender_blocked_as_slash32() { - let blocks = aggregate_v4(&[[9, 9, 9, 9]], 4, 8); - assert_eq!( - blocks, - vec![CidrV4 { - prefix_len: 32, - network: [9, 9, 9, 9] - }] - ); - } - - #[test] - fn four_sources_in_a_24_collapse_to_24() { - let srcs = [[10, 0, 5, 1], [10, 0, 5, 2], [10, 0, 5, 3], [10, 0, 5, 4]]; - let blocks = aggregate_v4(&srcs, 4, 8); - assert_eq!( - blocks, - vec![CidrV4 { - prefix_len: 24, - network: [10, 0, 5, 0] - }] - ); - } - - #[test] - fn three_sources_below_fanout_stay_as_32s() { - let srcs = [[10, 0, 5, 1], [10, 0, 5, 2], [10, 0, 5, 3]]; - let blocks = aggregate_v4(&srcs, 4, 8); - assert_eq!(blocks.len(), 3); - assert!(blocks.iter().all(|b| b.prefix_len == 32)); - } - - #[test] - fn collapses_multiple_levels_to_16() { - // Four /24s (each with four sources) under 10.0.x -> collapse to /16. - let mut srcs = Vec::new(); - for third in 0..4u8 { - for host in 1..=4u8 { - srcs.push([10, 0, third, host]); - } - } - // A permissive cap (/16) is required for this collapse. - let blocks = aggregate_v4(&srcs, 4, 16); - assert_eq!( - blocks, - vec![CidrV4 { - prefix_len: 16, - network: [10, 0, 0, 0] - }] - ); - } - - #[test] - fn default_cap_never_widens_past_slash24() { - // Same 16 sources spread across four /24s under 10.0.0.0/16 that would - // collapse to a /16 with a permissive cap must stay as four /24s under - // the safe default cap — never a /16 or wider. - let mut srcs = Vec::new(); - for third in 0..4u8 { - for host in 1..=4u8 { - srcs.push([10, 0, third, host]); - } - } - let blocks = aggregate_v4(&srcs, 4, DEFAULT_AGG_MAX_PREFIX_V4); - assert_eq!(blocks.len(), 4); - assert!( - blocks.iter().all(|b| b.prefix_len == 24), - "no block may be wider than /24: {blocks:?}" - ); - } - - #[test] - fn cap_prevents_slash8_blackhole() { - // Emulate the reported regression: many offenders scattered across a - // /8 (e.g. CDN edge IPs). With the default cap they must NEVER roll up - // into a /8 or /16 that blackholes the whole allocation. - let mut srcs = Vec::new(); - for second in 20..24u8 { - for third in 0..4u8 { - for host in 1..=4u8 { - srcs.push([104, second, third, host]); - } - } - } - let blocks = aggregate_v4(&srcs, 4, DEFAULT_AGG_MAX_PREFIX_V4); - assert!( - blocks.iter().all(|b| b.prefix_len >= 24), - "cap breached, produced a block wider than /24: {blocks:?}" - ); - } - - #[test] - fn distinct_24s_not_over_fanout_kept_separate() { - let srcs = [ - [10, 0, 5, 1], - [10, 0, 5, 2], - [10, 0, 5, 3], - [10, 0, 5, 4], // 10.0.5.0/24 - [10, 0, 9, 1], // lone source, different /24 - ]; - let blocks = aggregate_v4(&srcs, 4, 8); - assert!(blocks.contains(&CidrV4 { - prefix_len: 24, - network: [10, 0, 5, 0] - })); - assert!(blocks.contains(&CidrV4 { - prefix_len: 32, - network: [10, 0, 9, 1] - })); - assert_eq!(blocks.len(), 2); - } - - #[test] - fn plan_blocks_prefers_slash32_with_space() { - // Even a dense /24 worth of offenders stays as individual /32s while - // there is trie budget — no eager aggregation. - let srcs: Vec<[u8; 4]> = (1..=8).map(|h| [10, 0, 5, h]).collect(); - let blocks = plan_blocks_v4(&srcs, 4, 50_000, 24); - assert_eq!(blocks.len(), 8); - assert!(blocks.iter().all(|b| b.prefix_len == 32)); - } - - #[test] - fn plan_blocks_aggregates_only_under_pressure() { - // 8 offenders across two dense /24s, but a budget of only 2 entries - // forces aggregation — collapsing each /24. - let mut srcs: Vec<[u8; 4]> = (1..=4).map(|h| [10, 0, 5, h]).collect(); - srcs.extend((1..=4).map(|h| [10, 0, 6, h])); - let blocks = plan_blocks_v4(&srcs, 4, 2, 24); - assert_eq!(blocks.len(), 2); - assert!(blocks.iter().all(|b| b.prefix_len == 24)); - } - - #[test] - fn plan_blocks_never_exceeds_max_prefix_cap() { - // Under extreme pressure the planner still never goes wider than the cap. - let mut srcs = Vec::new(); - for third in 0..8u8 { - for host in 1..=4u8 { - srcs.push([10, 0, third, host]); - } - } - let blocks = plan_blocks_v4(&srcs, 4, 1, 24); - assert!( - blocks.iter().all(|b| b.prefix_len >= 24), - "cap breached: {blocks:?}" - ); + fn mask_v4_clears_host_bits() { + assert_eq!(mask_v4([10, 1, 2, 3], 24), [10, 1, 2, 0]); + assert_eq!(mask_v4([10, 1, 2, 3], 32), [10, 1, 2, 3]); + assert_eq!(mask_v4([10, 1, 2, 3], 0), [0, 0, 0, 0]); + assert_eq!(mask_v4([255, 255, 255, 255], 20), [255, 255, 240, 0]); } #[test] - fn v6_four_sources_in_64_collapse() { - let base = [0x20u8, 1, 0xd, 0xb8, 0, 0, 0, 0]; - let mk = |last: u8| { - let mut a = [0u8; 16]; - a[..8].copy_from_slice(&base); - a[15] = last; - a - }; - let srcs: Vec<[u8; 16]> = (1..=4).map(mk).collect(); - let blocks = aggregate_v6(&srcs, 4, 32); - assert_eq!(blocks.len(), 1); - assert_eq!(blocks[0].prefix_len, 64); - assert_eq!(&blocks[0].network[..8], &base); + fn mask_v6_clears_host_bits() { + let mut a = [0xffu8; 16]; + a[0] = 0x20; + let m = mask_v6(a, 48); + assert_eq!(&m[..6], &a[..6]); + assert!(m[6..].iter().all(|&b| b == 0)); } } diff --git a/lnvps_fw/lnvps_fw_service/src/config.rs b/lnvps_fw/lnvps_fw_service/src/config.rs index 2834c791..f0364104 100644 --- a/lnvps_fw/lnvps_fw_service/src/config.rs +++ b/lnvps_fw/lnvps_fw_service/src/config.rs @@ -313,43 +313,36 @@ pub struct Escalation { /// legitimately pushes hundreds–thousands of pps at a single origin. #[serde(default = "default_src_rate_pps")] pub src_rate_pps: u64, - /// Aggregation fan-out: this many child prefixes under a common parent - /// collapse into one wider block (/32->/24->/16->/8), keeping the LPM trie - /// bounded under distributed floods. + /// How long the kernel rate machine blocks a tripped source before it is + /// re-evaluated (re-extended each window it stays over-rate). + #[serde(default = "default_src_cooldown_secs")] + pub src_cooldown_secs: u64, + /// DEPRECATED, ignored: the per-source rate machine moved into the XDP + /// datapath, which blocks exact /32s|/128s in its state map — there is no + /// CIDR aggregation, spoof gate, or exit-hysteresis percentage anymore. + /// Kept parseable so existing config files keep loading across upgrades. #[serde(default = "default_agg_fanout")] pub agg_fanout: usize, - /// Widest IPv4 source block aggregation may ever produce (smallest prefix - /// length). Default /24 so a scatter of offenders (e.g. CDN edge IPs) can - /// never collapse into an allocation-crossing /16 or /8. + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_agg_max_prefix_v4")] pub agg_max_prefix_v4: u32, - /// Widest IPv6 source block aggregation may ever produce. Default /48. + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_agg_max_prefix_v6")] pub agg_max_prefix_v6: u32, - /// A DROPPING source's exit hysteresis: it returns to NORMAL (unblocked) - /// once its rate falls below this percentage of `src_rate_pps`. + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_exit_pct")] pub src_exit_pct: u64, - /// Sustained seconds below the source exit threshold before a source is - /// unblocked (hysteresis, mirrors the destination cooldown). - #[serde(default = "default_src_cooldown_secs")] - pub src_cooldown_secs: u64, - /// Block sources as individual /32s (/128s) up to this many entries per - /// family; only aggregate into wider prefixes when the trie would exceed - /// this budget ("merge only when running out of space"). + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_max_source_blocks")] pub max_source_blocks: usize, - /// A CIDR block is lifted after this many seconds without being refreshed - /// (safety upper-bound; the per-source state machine is the primary - /// release path). + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_block_ttl_secs")] pub block_ttl_secs: u64, /// Escalate a mitigating dest/prefix to source blocking only if this many /// packets/second are still getting through after the port-filter layer. #[serde(default = "default_escalate_pass_pps")] pub escalate_pass_pps: u64, - /// Spoof gate: if more than this many distinct offenders appear in a window - /// the flood is treated as spoofed and source blocking is skipped. + /// DEPRECATED, ignored (see `agg_fanout`). #[serde(default = "default_max_real_sources")] pub max_real_sources: usize, /// Enable the SYN_PROXY flag on a mitigating dest/prefix once its SYN rate @@ -537,15 +530,8 @@ impl Config { manual_v4: Vec::new(), manual_v6: Vec::new(), src_rate_pps: self.escalation.src_rate_pps, - fanout: self.escalation.agg_fanout, - agg_max_prefix_v4: self.escalation.agg_max_prefix_v4, - agg_max_prefix_v6: self.escalation.agg_max_prefix_v6, - src_exit_pct: self.escalation.src_exit_pct, src_cooldown_ns: self.escalation.src_cooldown_secs * 1_000_000_000, - max_source_blocks: self.escalation.max_source_blocks, - block_ttl_ns: self.escalation.block_ttl_secs * 1_000_000_000, escalate_pass_pps: self.escalation.escalate_pass_pps, - max_real_sources: self.escalation.max_real_sources, syn_proxy_pps: self.escalation.syn_proxy_syn_pps, }) } @@ -555,6 +541,26 @@ impl Config { mod tests { use super::*; + /// Config files written for the pre-in-kernel rate machine (with the + /// now-deprecated aggregation/spoof-gate/exit-hysteresis keys) MUST keep + /// parsing: a failed parse after a self-upgrade restart would take the + /// firewall down. The deprecated values are simply ignored. + #[test] + fn parses_legacy_escalation_keys() { + let cfg: Config = serde_yaml_ng::from_str( + "interfaces: [eno1]\nescalation:\n src-rate-pps: 500\n agg-fanout: 4\n \ + agg-max-prefix-v4: 24\n agg-max-prefix-v6: 48\n src-exit-pct: 50\n \ + src-cooldown-secs: 10\n max-source-blocks: 50000\n block-ttl-secs: 300\n \ + escalate-pass-pps: 50000\n max-real-sources: 10000\n syn-proxy-syn-pps: 5000\n", + ) + .unwrap(); + let rt = cfg.runtime_config().unwrap(); + assert_eq!(rt.src_rate_pps, 500); + assert_eq!(rt.src_cooldown_ns, 10_000_000_000); + assert_eq!(rt.escalate_pass_pps, 50_000); + assert_eq!(rt.syn_proxy_pps, 5_000); + } + #[test] fn parses_minimal_config() { let cfg: Config = serde_yaml_ng::from_str("interfaces: [eno1, eno2]\n").unwrap(); diff --git a/lnvps_fw/lnvps_fw_service/src/detect.rs b/lnvps_fw/lnvps_fw_service/src/detect.rs index a22fafaa..d0ff4243 100644 --- a/lnvps_fw/lnvps_fw_service/src/detect.rs +++ b/lnvps_fw/lnvps_fw_service/src/detect.rs @@ -197,105 +197,10 @@ pub fn process_sample( (transition, rates) } -// --- Per-source rate state machine --------------------------------------- -// -// A blocked source is not a sticky list entry: it has its own NORMAL <-> -// DROPPING state driven by its *current* packet rate, with the same entry -// threshold / exit hysteresis / cooldown shape as a destination. A source is -// only in DROPPING (and thus in the CIDR block trie) while it is actually over -// the rate limit; once it falls back below the exit threshold for the cooldown -// it returns to NORMAL and is unblocked — so legitimate low-rate traffic from -// an address that had one burst is not dropped for the whole TTL. - -/// Detection parameters for the per-source state machine. -#[derive(Debug, Clone, Copy)] -pub struct SourceDetectionConfig { - /// Enter DROPPING at or above this packets/second. - pub rate_pps: u64, - /// Exit hysteresis: leave DROPPING only once the rate falls below this - /// percentage of `rate_pps`. - pub exit_pct: u64, - /// Sustained time below the exit threshold before returning to NORMAL. - pub cooldown_ns: u64, -} - -/// Per-source userspace bookkeeping across sample windows. -#[derive(Debug, Clone, Copy, Default)] -pub struct SourceTracker { - /// False until the first observation has seeded `prev`. The eBPF source - /// counter is monotonic-cumulative and is not reset between mitigation - /// episodes, so a fresh tracker's first sample must establish the baseline - /// rather than treat the entire historical count as one window's traffic - /// (which would spike a benign source straight into DROPPING). - pub seeded: bool, - /// Previous cumulative packet count (for the rate delta). - pub prev: u64, - /// True while this source is being dropped (in the CIDR block trie). - pub dropping: bool, - /// Monotonic timestamp when the rate first fell below the exit threshold - /// (cooldown progress); `None` while above it. - pub below_since_ns: Option, - /// Most recent per-window rate (for display + block pps). - pub last_pps: u64, - /// Injected timestamp when `last_pps` was computed. - pub last_ns: u64, -} - -/// Advance one source's state machine from its fresh cumulative packet count. -/// Returns `(is_dropping, pps)` for this window. A counter that appears to have -/// decreased (LRU eviction / reset) yields a zero-rate window (never a spike). -pub fn advance_source( - tracker: &mut SourceTracker, - cur_count: u64, - cfg: &SourceDetectionConfig, - now_ns: u64, - elapsed_ns: u64, -) -> (bool, u64) { - // First observation: seed the baseline and report a zero-rate window. A - // single cumulative snapshot carries no rate information, and the counter - // may already hold a large historical total (see `seeded`). - if !tracker.seeded { - tracker.seeded = true; - tracker.prev = cur_count; - tracker.last_pps = 0; - tracker.last_ns = now_ns; - return (false, 0); - } - let pps = if elapsed_ns == 0 { - 0 - } else { - let delta = cur_count.saturating_sub(tracker.prev); - ((delta as u128 * 1_000_000_000u128) / elapsed_ns as u128) as u64 - }; - tracker.prev = cur_count; - tracker.last_pps = pps; - tracker.last_ns = now_ns; - - let exit = cfg.rate_pps.saturating_mul(cfg.exit_pct) / 100; - if !tracker.dropping { - if pps >= cfg.rate_pps { - tracker.dropping = true; - tracker.below_since_ns = None; - } - } else if pps < exit { - let since = *tracker.below_since_ns.get_or_insert(now_ns); - if now_ns.saturating_sub(since) >= cfg.cooldown_ns { - tracker.dropping = false; - tracker.below_since_ns = None; - } - } else { - // Rate climbed back up; restart the cooldown clock. - tracker.below_since_ns = None; - } - (tracker.dropping, pps) -} - #[cfg(test)] mod tests { use super::*; - const SEC: u64 = 1_000_000_000; - const CFG: DetectionConfig = DetectionConfig { pps: 1_000, syn_pps: 500, @@ -517,147 +422,4 @@ mod tests { assert_eq!(t, Transition::None); assert_eq!(below, None); } - - const SCFG: SourceDetectionConfig = SourceDetectionConfig { - rate_pps: 500, - exit_pct: 50, // exit below 250pps - cooldown_ns: 2_000_000_000, - }; - - /// Seed a fresh tracker's baseline at `base` cumulative packets (the first - /// observation always reports zero rate), so the following windows exercise - /// real deltas. - fn seeded_at(base: u64) -> SourceTracker { - let mut t = SourceTracker::default(); - let (dropping, pps) = advance_source(&mut t, base, &SCFG, 0, SEC); - assert!(!dropping); - assert_eq!(pps, 0, "first observation seeds, never spikes"); - t - } - - #[test] - fn source_enters_dropping_over_rate() { - let mut t = seeded_at(0); - // 600 pkts in the next 1s => 600pps >= 500 => DROPPING. - let (dropping, pps) = advance_source(&mut t, 600, &SCFG, SEC, SEC); - assert!(dropping); - assert_eq!(pps, 600); - } - - #[test] - fn source_stays_normal_below_rate() { - let mut t = seeded_at(0); - let (dropping, pps) = advance_source(&mut t, 100, &SCFG, SEC, SEC); - assert!(!dropping); - assert_eq!(pps, 100); - } - - #[test] - fn source_releases_after_cooldown_below_exit() { - let mut t = seeded_at(0); - // Enter DROPPING at 600pps. - advance_source(&mut t, 600, &SCFG, SEC, SEC); - assert!(t.dropping); - // Next window: no new packets (delta 0 => 0pps < 250 exit). Records - // "below since", still dropping (cooldown not elapsed). - let (d1, _) = advance_source(&mut t, 600, &SCFG, 2 * SEC, SEC); - assert!(d1, "still dropping during cooldown"); - // A full cooldown later: returns to NORMAL. - let (d2, _) = advance_source(&mut t, 600, &SCFG, 4 * SEC, SEC); - assert!(!d2, "released once below exit for the cooldown"); - } - - #[test] - fn source_resurgence_resets_cooldown() { - let mut t = seeded_at(0); - advance_source(&mut t, 600, &SCFG, SEC, SEC); - // Drop below exit (records below_since). - advance_source(&mut t, 600, &SCFG, 2 * SEC, SEC); - assert!(t.below_since_ns.is_some()); - // Rate climbs back over the entry threshold: cooldown clock clears. - let (d, _) = advance_source(&mut t, 1200, &SCFG, 3 * SEC, SEC); - assert!(d); - assert_eq!(t.below_since_ns, None); - } - - #[test] - fn source_counter_reset_is_zero_rate() { - let mut t = seeded_at(0); - advance_source(&mut t, 1000, &SCFG, SEC, SEC); - // Counter appears to decrease (LRU eviction/reset): zero rate, not a spike. - let (_, pps) = advance_source(&mut t, 10, &SCFG, 2 * SEC, SEC); - assert_eq!(pps, 0); - } - - // Regression: the eBPF `V*_SRC_COUNTERS` map is monotonic-cumulative and is - // NOT reset between mitigation episodes (LRU eviction is the only way an - // entry disappears). A source's userspace `SourceTracker`, however, is - // recreated fresh (`prev = 0`) each time it is first observed in a sample. - // - // So when mitigation (re)activates over a source whose counter already holds - // a large accumulated total from earlier counting, the very first window - // computes `delta = cur_count - 0 = cur_count` — the entire historical count - // attributed to a single window. A slow, benign source is then reported at a - // huge bogus pps and shoved straight into DROPPING even though its real rate - // is far below the limit. This is exactly the "manual activation puts IPs - // straight into dropping" symptom. - // - // Desired behaviour: the first observation must SEED the baseline, not emit a - // rate. This test asserts that and currently FAILS, demonstrating the bug. - #[test] - fn source_first_observation_of_large_counter_is_not_a_spike() { - let mut t = SourceTracker::default(); - // Counter already sits at 100_000 packets accumulated before this tracker - // ever existed (prior mitigation episode / pre-first-sample counting). - // The source is actually slow: it will add only ~5 pkts/s from here. - let (dropping, pps) = advance_source(&mut t, 100_000, &SCFG, 0, SEC); - assert!( - !dropping, - "benign source wrongly entered DROPPING on its first observation \ - (reported {pps} pps from a cumulative counter, limit {} pps)", - SCFG.rate_pps - ); - - // And its true rate (5 pkts over the next second) stays well under limit. - let (dropping2, pps2) = advance_source(&mut t, 100_005, &SCFG, 2 * SEC, SEC); - assert!(!dropping2, "slow source must stay NORMAL"); - assert_eq!(pps2, 5); - } - - // DEMONSTRATION (documents current behaviour, not yet the desired one): - // the per-source rate is evaluated over a single ~500ms sample window and a - // source is blocked on the FIRST window that reaches `rate_pps` — there is - // no entry hysteresis. So a source whose SUSTAINED rate is far below the - // limit is blocked by one short traffic burst (a CDN/reverse-proxy edge - // routinely bursts: connection storms, HTTP/2 multiplexing, retries). - // - // This test PASSES against the current code, proving the mechanism that - // puts low-average-rate Cloudflare edges into DROPPING even though they - // never exceeded the limit on a sustained basis. - #[test] - fn source_blocked_by_single_burst_window_despite_low_average() { - const HALF: u64 = SEC / 2; // 500ms sample window - let mut t = seeded_at(0); - - // Window 1 (500ms): a burst of 300 packets. Over a 500ms window that - // scales to 600pps, which is >= the 500pps limit. - let (dropping, pps) = advance_source(&mut t, 300, &SCFG, HALF, HALF); - assert_eq!(pps, 600, "a 300-packet burst in 500ms measures as 600pps"); - assert!( - dropping, - "current code blocks on the FIRST over-rate window (no entry hysteresis)" - ); - - // Yet the source's SUSTAINED average is well under the limit: over the - // full second (300 packets total, then quiet) it is only ~300pps, and - // over longer it trends toward its true low rate. It never sustained - // 500pps — a single half-second burst was enough to blackhole it. - let sustained_pps_over_1s = 300 * 1_000_000_000 / SEC; // 300 packets / 1s - assert!( - sustained_pps_over_1s < SCFG.rate_pps, - "sustained rate ({sustained_pps_over_1s}pps) is below the {}pps limit, \ - yet the source was blocked by one burst window", - SCFG.rate_pps - ); - } } diff --git a/lnvps_fw/lnvps_fw_service/src/gc.rs b/lnvps_fw/lnvps_fw_service/src/gc.rs index ff7edf18..737b8f8c 100644 --- a/lnvps_fw/lnvps_fw_service/src/gc.rs +++ b/lnvps_fw/lnvps_fw_service/src/gc.rs @@ -31,36 +31,40 @@ pub fn is_expired(last_seen_ns: u64, now_ns: u64, ttl_ns: u64) -> bool { now_ns.saturating_sub(last_seen_ns) > ttl_ns } -/// Sweep one learned-ports map, removing entries older than `ttl_ns`. Returns -/// the number of entries removed. -pub fn gc_open_ports( - map: &mut HashMap<&mut MapData, K, LastSeen>, +/// Select the expired keys from a pre-scanned learned-ports snapshot (pure — +/// the scan itself is done batched by the caller, so the sweep costs one +/// syscall per ~4k entries instead of two per entry). +pub fn expired_ports( + entries: &[(K, LastSeen)], now_ns: u64, tcp_ttl_ns: u64, udp_ttl_ns: u64, proto_of: impl Fn(&K) -> u8, -) -> usize +) -> Vec where K: Pod, { - // Collect keys first so the immutable iterator borrow is released before - // the mutable removals. - let keys: Vec = map.keys().flatten().collect(); - let mut removed = 0; - for k in keys { - let ttl_ns = if proto_of(&k) == lnvps_fw_common::PROTO_UDP { - udp_ttl_ns - } else { - tcp_ttl_ns - }; - if let Ok(v) = map.get(&k, 0) - && is_expired(v.last_seen, now_ns, ttl_ns) - && map.remove(&k).is_ok() - { - removed += 1; - } - } - removed + entries + .iter() + .filter(|(k, v)| { + let ttl_ns = if proto_of(k) == lnvps_fw_common::PROTO_UDP { + udp_ttl_ns + } else { + tcp_ttl_ns + }; + is_expired(v.last_seen, now_ns, ttl_ns) + }) + .map(|(k, _)| *k) + .collect() +} + +/// Remove a set of keys from a map, returning how many were actually removed +/// (an entry may have been LRU-evicted between the scan and the removal). +pub fn remove_keys(map: &mut HashMap<&mut MapData, K, LastSeen>, keys: &[K]) -> usize +where + K: Pod, +{ + keys.iter().filter(|k| map.remove(k).is_ok()).count() } #[cfg(test)] diff --git a/lnvps_fw/lnvps_fw_service/src/lib.rs b/lnvps_fw/lnvps_fw_service/src/lib.rs index 42310b22..53e1a118 100644 --- a/lnvps_fw/lnvps_fw_service/src/lib.rs +++ b/lnvps_fw/lnvps_fw_service/src/lib.rs @@ -3,6 +3,7 @@ //! (config parsing, learned-port GC) unit-testable and reusable. pub mod api; +pub mod batch; pub mod cidr; pub mod config; pub mod detect; diff --git a/lnvps_fw/lnvps_fw_service/src/main.rs b/lnvps_fw/lnvps_fw_service/src/main.rs index ccb957de..cbdc4746 100644 --- a/lnvps_fw/lnvps_fw_service/src/main.rs +++ b/lnvps_fw/lnvps_fw_service/src/main.rs @@ -25,7 +25,11 @@ use lnvps_fw_service::config::{Config, IfaceRole}; use lnvps_fw_service::detect::{DestTracker, DetectionConfig, Rates}; use lnvps_fw_service::gc; use lnvps_fw_service::publish::{MitInput, MitTracker}; -use lnvps_fw_service::runtime::{DetectionState, RuntimeConfig, run_control}; +use lnvps_fw_service::runtime::{self, DetectionState, RuntimeConfig, run_control, scan_plain}; + +/// A NORMAL (unblocked) source state idle for this long is swept from the +/// kernel state maps by the GC timer. +const SRC_STATE_IDLE_TTL_NS: u64 = 60 * 1_000_000_000; /// Sweep both learned-ports maps, returning the total number of entries /// removed. TTL is compared against the monotonic clock (matching @@ -34,18 +38,24 @@ fn gc_learned_ports(bpf: &mut Ebpf, tcp_ttl_ns: u64, udp_ttl_ns: u64) -> Result< let now = gc::monotonic_now_ns(); let mut removed = 0; { + // Batched scan first (one syscall per ~4k entries), then remove the + // expired keys through the typed handle. + let entries: Vec<(PortKeyV4, LastSeen)> = scan_plain(&*bpf, "OPEN_PORTS_V4")?; + let expired = gc::expired_ports(&entries, now, tcp_ttl_ns, udp_ttl_ns, |k| k.proto); let mut v4: AyaHashMap<_, PortKeyV4, LastSeen> = AyaHashMap::try_from( bpf.map_mut("OPEN_PORTS_V4") .context("open ports v4 missing")?, )?; - removed += gc::gc_open_ports(&mut v4, now, tcp_ttl_ns, udp_ttl_ns, |k| k.proto); + removed += gc::remove_keys(&mut v4, &expired); } { + let entries: Vec<(PortKeyV6, LastSeen)> = scan_plain(&*bpf, "OPEN_PORTS_V6")?; + let expired = gc::expired_ports(&entries, now, tcp_ttl_ns, udp_ttl_ns, |k| k.proto); let mut v6: AyaHashMap<_, PortKeyV6, LastSeen> = AyaHashMap::try_from( bpf.map_mut("OPEN_PORTS_V6") .context("open ports v6 missing")?, )?; - removed += gc::gc_open_ports(&mut v6, now, tcp_ttl_ns, udp_ttl_ns, |k| k.proto); + removed += gc::remove_keys(&mut v6, &expired); } Ok(removed) } @@ -162,18 +172,17 @@ where K: aya::Pod + Eq + std::hash::Hash, { let now = gc::monotonic_now_ns(); + // Batched scan (one syscall per ~4k entries), then targeted removals. + let entries: Vec<(K, u64)> = scan_plain(&*bpf, name)?; + let expired: Vec = entries + .iter() + .filter(|(_, ts)| gc::is_expired(*ts, now, ttl_ns)) + .map(|(k, _)| *k) + .collect(); let mut map: AyaHashMap<_, K, u64> = AyaHashMap::try_from( bpf.map_mut(name) .with_context(|| format!("{name} missing"))?, )?; - let expired: Vec = map - .keys() - .flatten() - .filter(|k| match map.get(k, 0) { - Ok(ts) => gc::is_expired(ts, now, ttl_ns), - Err(_) => false, - }) - .collect(); let mut removed = 0; for k in &expired { if map.remove(k).is_ok() { @@ -354,79 +363,76 @@ fn collect_tracked( out } -/// Snapshot the active blocked source CIDRs (from SOURCE_BLOCK escalation). +/// Estimate a source's live pps from its kernel window state: packets counted +/// this window over the window's elapsed time. An idle source's stale window +/// decays toward zero naturally (elapsed keeps growing while count doesn't). +/// Elapsed is floored at 100ms so a freshly-rolled window can't inflate. +fn src_pps_estimate(st: &lnvps_fw_common::SrcState, now_ns: u64) -> u64 { + let elapsed = now_ns.saturating_sub(st.window_start_ns).max(100_000_000); + ((st.count as u128 * 1_000_000_000u128) / elapsed as u128) as u64 +} + +/// Snapshot the currently-blocked sources (kernel `blocked_until` in the +/// future) as /32|/128 "blocks" for the legacy `/blocks` view. The block list +/// IS the per-source state map now — there is no CIDR aggregation; `age_secs` +/// reports seconds until the block expires (re-extended while still over-rate) +/// and `cooling` is always false (the exit-hysteresis display state is gone). fn collect_blocks(det: &DetectionState, now_ns: u64) -> Vec { - let age = |ts: u64| now_ns.saturating_sub(ts) / 1_000_000_000; - // A block is "cooling" if none of its covered sources is still actively - // over-rate (all have entered the exit-hysteresis countdown). - let cooling_v4 = |c: &lnvps_fw_service::cidr::CidrV4| { - !det.src_v4.iter().any(|(ip, t)| { - t.dropping && t.below_since_ns.is_none() && mask_v4(*ip, c.prefix_len) == c.network - }) - }; - let cooling_v6 = |c: &lnvps_fw_service::cidr::CidrV6| { - !det.src_v6.iter().any(|(ip, t)| { - t.dropping && t.below_since_ns.is_none() && mask_v6(*ip, c.prefix_len) == c.network - }) - }; + let remaining = |until: u64| until.saturating_sub(now_ns) / 1_000_000_000; let mut out: Vec = det - .blocks_v4 + .src_v4 .iter() - .map(|(c, &ts)| SourceBlock { - cidr: format!("{}/{}", Ipv4Addr::from(c.network), c.prefix_len), - age_secs: age(ts), - pps: det.block_pps_v4.get(c).copied().unwrap_or(0), + .filter(|(_, st)| st.blocked_until_ns > now_ns) + .map(|(ip, st)| SourceBlock { + cidr: format!("{}/32", Ipv4Addr::from(*ip)), + age_secs: remaining(st.blocked_until_ns), + pps: src_pps_estimate(st, now_ns), manual: false, - cooling: cooling_v4(c), + cooling: false, }) - .chain(det.blocks_v6.iter().map(|(c, &ts)| SourceBlock { - cidr: format!("{}/{}", Ipv6Addr::from(c.network), c.prefix_len), - age_secs: age(ts), - pps: det.block_pps_v6.get(c).copied().unwrap_or(0), - manual: false, - cooling: cooling_v6(c), - })) + .chain( + det.src_v6 + .iter() + .filter(|(_, st)| st.blocked_until_ns > now_ns) + .map(|(ip, st)| SourceBlock { + cidr: format!("{}/128", Ipv6Addr::from(*ip)), + age_secs: remaining(st.blocked_until_ns), + pps: src_pps_estimate(st, now_ns), + manual: false, + cooling: false, + }), + ) .collect(); // Most active first (the API re-sorts the merged manual+auto set too). out.sort_by(|a, b| b.pps.cmp(&a.pps).then_with(|| a.cidr.cmp(&b.cidr))); out } -/// Snapshot every rate-tracked source (all states) for the `/sources` view. -/// While a destination is mitigating the eBPF counts each source and the -/// per-source state machine lives in `det.src_v4`/`src_v6`; this exposes the -/// whole set with its 3-state label so the UI can show NORMAL sources too, not -/// just the blocked (dropping/cooling) subset that `/blocks` carries. +/// Snapshot every kernel-tracked source for the `/sources` view. The rate +/// machine (and the blocking decision) lives in XDP; this only relabels its +/// state for display: blocked → `dropping`, otherwise `normal`. fn collect_sources(det: &DetectionState, now_ns: u64) -> Vec { - let age = |ns: u64| now_ns.saturating_sub(ns) / 1_000_000_000; - // NORMAL = not dropping; DROPPING = at/over rate (no cooldown clock yet); - // COOLING = still dropping but below the exit threshold, counting down. - let state = |t: &lnvps_fw_service::detect::SourceTracker| { - if !t.dropping { - "normal" - } else if t.below_since_ns.is_none() { + let row = |ip: String, st: &lnvps_fw_common::SrcState| TrackedSource { + ip, + pps: src_pps_estimate(st, now_ns), + state: if st.blocked_until_ns > now_ns { "dropping" } else { - "cooling" + "normal" } + .to_string(), + manual: false, + age_secs: now_ns.saturating_sub(st.window_start_ns) / 1_000_000_000, }; let mut out: Vec = det .src_v4 .iter() - .map(|(ip, t)| TrackedSource { - ip: Ipv4Addr::from(*ip).to_string(), - pps: t.last_pps, - state: state(t).to_string(), - manual: false, - age_secs: age(t.last_ns), - }) - .chain(det.src_v6.iter().map(|(ip, t)| TrackedSource { - ip: Ipv6Addr::from(*ip).to_string(), - pps: t.last_pps, - state: state(t).to_string(), - manual: false, - age_secs: age(t.last_ns), - })) + .map(|(ip, st)| row(Ipv4Addr::from(*ip).to_string(), st)) + .chain( + det.src_v6 + .iter() + .map(|(ip, st)| row(Ipv6Addr::from(*ip).to_string(), st)), + ) .collect(); // Most active first (the API re-sorts + paginates too). out.sort_by(|a, b| b.pps.cmp(&a.pps).then_with(|| a.ip.cmp(&b.ip))); @@ -448,7 +454,6 @@ fn apply_limits(rt: &mut RuntimeConfig, l: &Limits) { rt.network.exit_pct = l.exit_pct; rt.network.cooldown_ns = cooldown_ns; rt.src_rate_pps = l.src_rate_pps; - rt.src_exit_pct = l.src_exit_pct; rt.src_cooldown_ns = l.src_cooldown_secs.saturating_mul(1_000_000_000); } @@ -1069,10 +1074,11 @@ async fn main() -> Result<()> { exit_pct: det.exit_pct, cooldown_secs: det.cooldown_ns / 1_000_000_000, src_rate_pps: runtime_cfg.src_rate_pps, - src_exit_pct: runtime_cfg.src_exit_pct, src_cooldown_secs: runtime_cfg.src_cooldown_ns / 1_000_000_000, }); } + // Arm the in-kernel per-source rate machine before traffic decisions. + runtime::write_src_rate_cfg(&mut bpf, &runtime_cfg)?; let mut detect_timer = tokio::time::interval(cfg.sample_interval()); let mut gc_timer = tokio::time::interval(cfg.gc_interval()); // Rotate the SYN-cookie secret periodically; cookies issued in the previous @@ -1133,6 +1139,11 @@ async fn main() -> Result<()> { if v != limits_version { limits_version = v; apply_limits(&mut runtime_cfg, &st.limits()); + // Push the per-source limits into the kernel rate + // machine (it owns per-source blocking now). + if let Err(e) = runtime::write_src_rate_cfg(&mut bpf, &runtime_cfg) { + warn!("writing src rate config failed: {e}"); + } info!("detection limits updated via API"); } } @@ -1229,6 +1240,14 @@ async fn main() -> Result<()> { if let Err(e) = gc_verified(&mut bpf, verified_ttl_ns) { warn!("verified GC failed: {e}"); } + // Sweep idle per-source rate states (the kernel machine never + // self-cleans; without this the state maps and the /sources + // view would hold every source ever seen under mitigation). + match runtime::gc_src_states(&mut bpf, SRC_STATE_IDLE_TTL_NS) { + Ok(n) if n > 0 => info!("GC removed {n} idle source state(s)"), + Ok(_) => {} + Err(e) => warn!("source-state GC failed: {e}"), + } // Refresh the learned-ports snapshot for the control API. if let Some(st) = &api_state { st.set_ports(collect_ports(&bpf)); @@ -1245,3 +1264,36 @@ async fn main() -> Result<()> { info!("Shutdown complete."); Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use lnvps_fw_common::SrcState; + + const SEC: u64 = 1_000_000_000; + + #[test] + fn src_pps_estimate_scales_count_over_window_elapsed() { + let st = SrcState { + window_start_ns: 0, + count: 500, + blocked_until_ns: 0, + }; + // Half a second into the window, 500 packets => 1000 pps. + assert_eq!(src_pps_estimate(&st, SEC / 2), 1000); + // A stale window decays naturally: same 500 packets over 10s => 50. + assert_eq!(src_pps_estimate(&st, 10 * SEC), 50); + } + + #[test] + fn src_pps_estimate_floors_tiny_elapsed() { + let st = SrcState { + window_start_ns: 0, + count: 10, + blocked_until_ns: 0, + }; + // 1ms into a fresh window: elapsed floored at 100ms so the estimate is + // 100 pps, not 10_000 (no burst inflation from a just-rolled window). + assert_eq!(src_pps_estimate(&st, 1_000_000), 100); + } +} diff --git a/lnvps_fw/lnvps_fw_service/src/runtime.rs b/lnvps_fw/lnvps_fw_service/src/runtime.rs index b7f9a0ba..0e729436 100644 --- a/lnvps_fw/lnvps_fw_service/src/runtime.rs +++ b/lnvps_fw/lnvps_fw_service/src/runtime.rs @@ -15,7 +15,7 @@ //! //! The eBPF side only counts and enforces; every threshold decision is here. -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::hash::Hash; use std::net::{Ipv4Addr, Ipv6Addr}; @@ -27,13 +27,12 @@ use log::{info, warn}; use lnvps_fw_common::{ DEST_MODE_NORMAL, DEST_MODE_PORT_FILTER, DEST_MODE_SOURCE_BLOCK, DEST_MODE_SYN_PROXY, - DestCounters, DestState, + DestCounters, DestState, SrcRateConfig, SrcState, }; -use crate::cidr::{CidrV4, CidrV6, mask_v4, mask_v6, plan_blocks_v4, plan_blocks_v6}; +use crate::cidr::{mask_v4, mask_v6}; use crate::detect::{ - DestTracker, DetectionConfig, Rates, SourceDetectionConfig, SourceTracker, Transition, - advance_source, compute_rates, process_sample, + DestTracker, DetectionConfig, Rates, Transition, compute_rates, process_sample, }; /// Runtime configuration for one control tick. @@ -56,36 +55,15 @@ pub struct RuntimeConfig { pub manual_v4: Vec<(u32, [u8; 4], u32)>, /// Manual override flags (IPv6). pub manual_v6: Vec<(u32, [u8; 16], u32)>, - /// Per-source packets/second that marks a source as an offender. + /// Per-source packets/second limit for the in-kernel rate machine + /// (written to `SRC_RATE_CFG` as `max_per_window` over a 1s window). pub src_rate_pps: u64, - /// Aggregation fan-out: this many child prefixes under a parent collapse to - /// the parent (/32->/24->/16->/8, /128->/64->/48->/32). - pub fanout: usize, - /// Widest IPv4 source block aggregation may ever produce (smallest prefix - /// length, e.g. 24 = never wider than a /24). Prevents a scatter of - /// offenders from collapsing into a huge allocation-crossing block. - pub agg_max_prefix_v4: u32, - /// Widest IPv6 source block aggregation may ever produce. - pub agg_max_prefix_v6: u32, - /// A DROPPING source's exit hysteresis (% of `src_rate_pps`). - pub src_exit_pct: u64, - /// Sustained time below the source exit threshold before a source returns - /// to NORMAL and is unblocked. + /// How long the kernel machine blocks a tripped source before it is + /// re-evaluated (re-extended each window it is still over-rate). pub src_cooldown_ns: u64, - /// Trie-space budget: block sources as individual /32s (v4) / /128s (v6) - /// until this many entries, only aggregating under pressure beyond it. - pub max_source_blocks: usize, - /// A CIDR block is lifted this many ns after it stops being refreshed - /// (safety upper-bound for sources evicted from the per-source counter LRU; - /// the per-source state machine is the primary release mechanism). - pub block_ttl_ns: u64, /// Escalate a mitigating dest/prefix to `SOURCE_BLOCK` only if this many /// packets/second are still getting through after the port filter. pub escalate_pass_pps: u64, - /// Spoof gate: if more than this many distinct offenders are seen in a - /// window, treat the flood as spoofed and skip source blocking entirely - /// (rely on the port filter instead of chasing unblockable /32s). - pub max_real_sources: usize, /// Enable the SYN_PROXY flag once a mitigating entity's SYN rate reaches /// this many SYNs/second. pub syn_proxy_pps: u64, @@ -122,18 +100,13 @@ pub struct DetectionState { /// Per-protected-prefix detection trackers, keyed by (prefix_len, network). pub prefix_v4: HashMap<(u32, [u8; 4]), DestTracker>, pub prefix_v6: HashMap<(u32, [u8; 16]), DestTracker>, - /// Per-source rate state machines (NORMAL/DROPPING with hysteresis), keyed - /// by source address. A source is in the block trie only while its tracker - /// is DROPPING; it is released as soon as its rate falls back (hysteresis), - /// not held for a blind TTL. - pub src_v4: HashMap<[u8; 4], SourceTracker>, - pub src_v6: HashMap<[u8; 16], SourceTracker>, - /// Active CIDR blocks -> timestamp last refreshed (for TTL decay). - pub blocks_v4: HashMap, - pub blocks_v6: HashMap, - /// Active CIDR blocks -> current aggregate pps from sources under them. - pub block_pps_v4: HashMap, - pub block_pps_v6: HashMap, + /// Latest batched snapshot of the kernel-owned per-source rate states + /// (display only — the rate machine and blocking decision live in XDP). + pub src_v4: Vec<([u8; 4], SrcState)>, + pub src_v6: Vec<([u8; 16], SrcState)>, + /// `bpf_ktime_get_ns`-domain timestamp of the snapshot (SrcState fields + /// are kernel-monotonic; compare against this, not userspace clocks). + pub src_sampled_ns: u64, /// Previous cumulative TX (egress) counter snapshots, keyed by local IP. pub prev_tx_v4: HashMap<[u8; 4], DestCounters>, pub prev_tx_v6: HashMap<[u8; 16], DestCounters>, @@ -146,17 +119,79 @@ pub struct DetectionState { pub fn sum_counters<'a>(values: impl IntoIterator) -> DestCounters { let mut total = DestCounters::default(); for v in values { - total.packets += v.packets; - total.bytes += v.bytes; - total.syn_packets += v.syn_packets; - total.tcp_packets += v.tcp_packets; - total.udp_packets += v.udp_packets; - total.icmp_packets += v.icmp_packets; - total.dropped += v.dropped; + add_counters(&mut total, v); } total } +/// Add one per-CPU `DestCounters` slot into an accumulator (batch-read fold). +fn add_counters(acc: &mut DestCounters, v: &DestCounters) { + acc.packets += v.packets; + acc.bytes += v.bytes; + acc.syn_packets += v.syn_packets; + acc.tcp_packets += v.tcp_packets; + acc.udp_packets += v.udp_packets; + acc.icmp_packets += v.icmp_packets; + acc.dropped += v.dropped; +} + +/// Cached count of possible CPUs (the per-CPU map value stride). +fn possible_cpus() -> Result { + static N: std::sync::OnceLock = std::sync::OnceLock::new(); + if let Some(n) = N.get() { + return Ok(*n); + } + let n = aya::util::nr_cpus().map_err(|(what, e)| anyhow::anyhow!("{what}: {e}"))?; + Ok(*N.get_or_init(|| n)) +} + +/// The map's fd if its type takes the hash-family `BPF_MAP_LOOKUP_BATCH` path. +fn batch_fd(map: &aya::maps::Map) -> Option> { + use std::os::fd::AsFd; + match map { + aya::maps::Map::HashMap(d) + | aya::maps::Map::LruHashMap(d) + | aya::maps::Map::PerCpuHashMap(d) + | aya::maps::Map::PerCpuLruHashMap(d) => Some(d.fd().as_fd()), + _ => None, + } +} + +/// Handle a batch-read failure: permanent unsupport logs once and returns +/// `Ok(())` so the caller falls through to per-entry iteration; anything else +/// propagates. +fn batch_fallback(name: &str, e: std::io::Error) -> Result<()> { + if crate::batch::note_failure(&e) { + warn!("batched read of {name} unsupported ({e}); using per-entry iteration"); + Ok(()) + } else { + Err(anyhow::anyhow!("batched read of {name} failed: {e}")) + } +} + +/// Batched read of a **plain** hash map with per-entry fallback (GC scans). +pub fn scan_plain(bpf: &Ebpf, name: &str) -> Result> +where + K: Pod, + V: Pod + Copy, +{ + let map = bpf.map(name).with_context(|| format!("{name} missing"))?; + if crate::batch::supported() + && let Some(fd) = batch_fd(map) + { + match crate::batch::read_plain::(fd) { + Ok(v) => return Ok(v), + Err(e) => batch_fallback(name, e)?, + } + } + let map: aya::maps::HashMap<_, K, V> = aya::maps::HashMap::try_from(map)?; + let mut out = Vec::new(); + for entry in map.iter() { + out.push(entry?); + } + Ok(out) +} + fn dest_state(level: u32, now_ns: u64) -> DestState { DestState { mode: level, @@ -170,19 +205,17 @@ fn dest_state(level: u32, now_ns: u64) -> DestState { /// (bounded/real offenders present) and traffic is still getting through after /// the port filter. Other flags (SYN_PROXY, RATE_CAPS) are OR'd in here as they /// are implemented, so any subset can be active at once. -fn enforced_flags( - rates: &Rates, - source_block_active: bool, - escalate_pass_pps: u64, - syn_proxy_pps: u64, -) -> u32 { +fn enforced_flags(rates: &Rates, escalate_pass_pps: u64, syn_proxy_pps: u64) -> u32 { let mut flags = DEST_MODE_PORT_FILTER; // A sustained SYN flood engages the SYN-proxy (validate handshakes to open // TCP ports with cookies). High efficacy vs spoofed SYN floods, low FP. if rates.syn_pps >= syn_proxy_pps { flags |= DEST_MODE_SYN_PROXY; } - if source_block_active && rates.pass_pps >= escalate_pass_pps { + // Escalate on residual pass-rate alone: the in-kernel gate only ever + // blocks sources genuinely over the per-source limit, so a spoofed flood + // (which never trips it) makes the flag harmless rather than dangerous. + if rates.pass_pps >= escalate_pass_pps { flags |= DEST_MODE_SOURCE_BLOCK; } flags @@ -211,7 +244,6 @@ fn detect_family( bits: u32, trackers: &mut HashMap, cfg: &DetectionConfig, - source_block_active: bool, escalate_pass_pps: u64, syn_proxy_pps: u64, now_ns: u64, @@ -244,12 +276,7 @@ fn detect_family( } // Active: pick the enforcement level (never below the manual floor) and // (re)write it if changed. - let target = enforced_flags( - &rates, - source_block_active, - escalate_pass_pps, - syn_proxy_pps, - ) | manual; + let target = enforced_flags(&rates, escalate_pass_pps, syn_proxy_pps) | manual; if transition == Transition::Entered { let _ = trie.insert(&Key::new(bits, *key), dest_state(target, now_ns), 0); tracker.flags = target; @@ -280,7 +307,6 @@ fn detect_prefix( mask: impl Fn(K, u32) -> K, trackers: &mut HashMap<(u32, K), DestTracker>, cfg: &DetectionConfig, - source_block_active: bool, escalate_pass_pps: u64, syn_proxy_pps: u64, now_ns: u64, @@ -321,12 +347,7 @@ fn detect_prefix( } return; } - let target = enforced_flags( - &rates, - source_block_active, - escalate_pass_pps, - syn_proxy_pps, - ) | manual; + let target = enforced_flags(&rates, escalate_pass_pps, syn_proxy_pps) | manual; if transition == Transition::Entered { let _ = trie.insert( &Key::new(prefix_len, network), @@ -362,231 +383,88 @@ fn detect_prefix( } } -/// Read + per-CPU-sum a `DestCounters` map into an owned vec. +/// Read + per-CPU-sum a `DestCounters` map into an owned vec. Uses one +/// `BPF_MAP_LOOKUP_BATCH` syscall per ~4k entries, falling back to aya's +/// per-entry iteration (2 syscalls/entry) on kernels without batch support. fn read_counters(bpf: &Ebpf, name: &str) -> Result> where K: Pod, { - let map: PerCpuHashMap<_, K, DestCounters> = - PerCpuHashMap::try_from(bpf.map(name).with_context(|| format!("{name} missing"))?)?; - let mut out = Vec::new(); - for entry in map.iter() { - let (k, values) = entry?; - out.push((k, sum_counters(values.iter()))); + let map = bpf.map(name).with_context(|| format!("{name} missing"))?; + if crate::batch::supported() + && let Some(fd) = batch_fd(map) + { + match crate::batch::read_percpu_folded::( + fd, + possible_cpus()?, + add_counters, + ) { + Ok(v) => return Ok(v), + Err(e) => batch_fallback(name, e)?, + } } - Ok(out) -} - -/// Read + per-CPU-sum a per-source `u64` counter map into an owned vec. -fn read_src_counters(bpf: &Ebpf, name: &str) -> Result> -where - K: Pod, -{ - let map: PerCpuHashMap<_, K, u64> = - PerCpuHashMap::try_from(bpf.map(name).with_context(|| format!("{name} missing"))?)?; + let map: PerCpuHashMap<_, K, DestCounters> = PerCpuHashMap::try_from(map)?; let mut out = Vec::new(); for entry in map.iter() { let (k, values) = entry?; - out.push((k, values.iter().copied().sum())); + out.push((k, sum_counters(values.iter()))); } Ok(out) } -/// Advance every source's rate state machine for one family and return the set -/// of addresses currently in DROPPING. Sources not sampled this window (evicted -/// from the counter LRU, or gone quiet) are driven toward NORMAL with a -/// zero-rate window and dropped from the tracker map once they return to -/// NORMAL, so the map stays bounded. -fn step_sources( - cur: &[(K, u64)], - trackers: &mut HashMap, - scfg: &SourceDetectionConfig, - now_ns: u64, - elapsed_ns: u64, -) -> Vec -where - K: Eq + Hash + Copy, -{ - let cur_map: HashMap = cur.iter().copied().collect(); - let mut dropping = Vec::new(); - for (k, c) in cur { - let t = trackers.entry(*k).or_default(); - let (drop, _) = advance_source(t, *c, scfg, now_ns, elapsed_ns); - if drop { - dropping.push(*k); - } - } - // Sources absent from this window's sample: no new packets, so advance them - // with a zero-rate window (delta 0). Once a source returns to NORMAL it is - // removed; while still cooling down in DROPPING it stays blocked. - let stale: Vec = trackers - .keys() - .filter(|k| !cur_map.contains_key(*k)) - .copied() - .collect(); - for k in stale { - let t = trackers.get_mut(&k).expect("stale key present"); - let prev = t.prev; - let (drop, _) = advance_source(t, prev, scfg, now_ns, elapsed_ns); - if drop { - dropping.push(k); - } else { - trackers.remove(&k); - } - } - dropping +/// Fixed per-source counting window for the in-kernel rate machine. One +/// second: `src_rate_pps` is then literally "packets per second", exact, not +/// a delta over a variable sample interval. +pub const SRC_WINDOW_NS: u64 = 1_000_000_000; + +/// Write the kernel per-source rate-machine config (`SRC_RATE_CFG[0]`). +/// Called at startup and whenever the limits change (`PUT /limits`). +/// `max_per_window` is precomputed so the datapath never divides. +pub fn write_src_rate_cfg(bpf: &mut Ebpf, cfg: &RuntimeConfig) -> Result<()> { + let mut arr: aya::maps::Array<_, SrcRateConfig> = aya::maps::Array::try_from( + bpf.map_mut("SRC_RATE_CFG") + .context("SRC_RATE_CFG missing")?, + )?; + let c = SrcRateConfig { + max_per_window: cfg.src_rate_pps.saturating_mul(SRC_WINDOW_NS) / 1_000_000_000, + window_ns: SRC_WINDOW_NS, + cooldown_ns: cfg.src_cooldown_ns, + }; + arr.set(0, c, 0).context("writing SRC_RATE_CFG")?; + Ok(()) } -/// Reconcile one family's CIDR block trie to exactly the `desired` set: install -/// entries that are newly desired, remove entries no longer desired. Unlike the -/// old TTL scheme this is a pure set-diff against the per-source state machine's -/// current DROPPING set — a source is unblocked the moment its tracker leaves -/// DROPPING, not after a blind TTL. -fn reconcile_block_set( - blocks: &mut HashMap, - trie: &mut LpmTrie<&mut MapData, K, u8>, - desired: &[C], - now_ns: u64, - key_of: impl Fn(&C) -> Key, - fmt_cidr: impl Fn(&C) -> String, -) where - K: Pod, - C: Eq + Hash + Copy, -{ - let want: HashSet = desired.iter().copied().collect(); - for c in desired { - if !blocks.contains_key(c) { - if let Err(e) = trie.insert(&key_of(c), 1, 0) { - warn!("failed to install CIDR block {}: {e}", fmt_cidr(c)); - continue; - } - blocks.insert(*c, now_ns); - warn!("CIDR BLOCK {}", fmt_cidr(c)); - } - } - let gone: Vec = blocks - .keys() - .filter(|c| !want.contains(c)) - .copied() - .collect(); - for c in gone { - let _ = trie.remove(&key_of(&c)); - blocks.remove(&c); - info!("CIDR UNBLOCK {}", fmt_cidr(&c)); - } +/// Remove idle per-source state entries: the kernel machine never self-cleans, +/// so without this sweep the state maps (and the `/sources` view) would hold +/// every source ever seen under mitigation. An entry is stale once it is not +/// blocked and its window anchor is older than `idle_ttl_ns`. Runs on the slow +/// GC timer — steady-state cost is proportional to live sources, not history. +pub fn gc_src_states(bpf: &mut Ebpf, idle_ttl_ns: u64) -> Result { + let now = crate::gc::monotonic_now_ns(); + let mut removed = 0; + removed += gc_src_state_map::<[u8; 4]>(bpf, "V4_SRC_STATE", now, idle_ttl_ns)?; + removed += gc_src_state_map::<[u8; 16]>(bpf, "V6_SRC_STATE", now, idle_ttl_ns)?; + Ok(removed) } -/// Aggregate per-block current pps from the per-source trackers under each -/// active block (for the API/dashboard view). -fn block_pps( - blocks: &HashMap, - trackers: &HashMap, - covers: impl Fn(&C, &K) -> bool, -) -> HashMap +fn gc_src_state_map(bpf: &mut Ebpf, name: &str, now_ns: u64, idle_ttl_ns: u64) -> Result where - K: Eq + Hash + Copy, - C: Eq + Hash + Copy, + K: Pod + Eq + Hash, { - blocks - .keys() - .map(|c| { - let sum = trackers - .iter() - .filter(|(ip, _)| covers(c, ip)) - .map(|(_, t)| t.last_pps) - // (covers takes &C, &K; ip is &&K via match ergonomics) - .sum(); - (*c, sum) + let entries: Vec<(K, SrcState)> = scan_plain(&*bpf, name)?; + let stale: Vec = entries + .iter() + .filter(|(_, st)| { + st.blocked_until_ns <= now_ns + && now_ns.saturating_sub(st.window_start_ns) >= idle_ttl_ns }) - .collect::>() -} - -fn src_cfg(cfg: &RuntimeConfig) -> SourceDetectionConfig { - SourceDetectionConfig { - rate_pps: cfg.src_rate_pps, - exit_pct: cfg.src_exit_pct, - cooldown_ns: cfg.src_cooldown_ns, - } -} - -/// Source analysis for one family: advance the per-source state machines, apply -/// the spoof gate, plan the block set (/32-first, aggregating only under trie -/// pressure), and reconcile the trie. Returns whether any block is active -/// (drives escalation to `SOURCE_BLOCK`). -fn source_control_v4( - bpf: &mut Ebpf, - state: &mut DetectionState, - cfg: &RuntimeConfig, - now_ns: u64, - elapsed_ns: u64, -) -> Result { - let cur = read_src_counters::<[u8; 4]>(bpf, "V4_SRC_COUNTERS")?; - let dropping = step_sources(&cur, &mut state.src_v4, &src_cfg(cfg), now_ns, elapsed_ns); - // Spoof gate: an unbounded DROPPING set means a spoofed flood — skip source - // blocking (chasing spoofed /32s is pointless; rely on the port filter). - let desired = if dropping.len() > cfg.max_real_sources { - Vec::new() - } else { - plan_blocks_v4( - &dropping, - cfg.fanout, - cfg.max_source_blocks, - cfg.agg_max_prefix_v4, - ) - }; - { - let mut trie: LpmTrie<_, [u8; 4], u8> = - LpmTrie::try_from(bpf.map_mut("V4_CIDR_SRC").context("v4 cidr trie missing")?)?; - reconcile_block_set( - &mut state.blocks_v4, - &mut trie, - &desired, - now_ns, - |c| Key::new(c.prefix_len, c.network), - |c| format!("{}/{}", Ipv4Addr::from(c.network), c.prefix_len), - ); - } - state.block_pps_v4 = block_pps(&state.blocks_v4, &state.src_v4, |c, ip| { - mask_v4(*ip, c.prefix_len) == c.network - }); - Ok(!state.blocks_v4.is_empty()) -} - -fn source_control_v6( - bpf: &mut Ebpf, - state: &mut DetectionState, - cfg: &RuntimeConfig, - now_ns: u64, - elapsed_ns: u64, -) -> Result { - let cur = read_src_counters::<[u8; 16]>(bpf, "V6_SRC_COUNTERS")?; - let dropping = step_sources(&cur, &mut state.src_v6, &src_cfg(cfg), now_ns, elapsed_ns); - let desired = if dropping.len() > cfg.max_real_sources { - Vec::new() - } else { - plan_blocks_v6( - &dropping, - cfg.fanout, - cfg.max_source_blocks, - cfg.agg_max_prefix_v6, - ) - }; - { - let mut trie: LpmTrie<_, [u8; 16], u8> = - LpmTrie::try_from(bpf.map_mut("V6_CIDR_SRC").context("v6 cidr trie missing")?)?; - reconcile_block_set( - &mut state.blocks_v6, - &mut trie, - &desired, - now_ns, - |c| Key::new(c.prefix_len, c.network), - |c| format!("{}/{}", Ipv6Addr::from(c.network), c.prefix_len), - ); - } - state.block_pps_v6 = block_pps(&state.blocks_v6, &state.src_v6, |c, ip| { - mask_v6(*ip, c.prefix_len) == c.network - }); - Ok(!state.blocks_v6.is_empty()) + .map(|(k, _)| *k) + .collect(); + let mut map: aya::maps::HashMap<_, K, SrcState> = aya::maps::HashMap::try_from( + bpf.map_mut(name) + .with_context(|| format!("{name} missing"))?, + )?; + Ok(stale.iter().filter(|k| map.remove(k).is_ok()).count()) } /// Sample one TX-counter map, compute per-local-IP egress rates against the @@ -628,9 +506,11 @@ pub fn run_control( }; state.last_sample_ns = now_ns; - // --- Source control first (spoof-gated); result gates escalation --- - let sba4 = source_control_v4(bpf, state, cfg, now_ns, elapsed)?; - let sba6 = source_control_v6(bpf, state, cfg, now_ns, elapsed)?; + // --- Per-source rate machine lives in XDP; snapshot its state maps for + // the display views only (batched: ~1 syscall per 4k entries) --- + state.src_v4 = scan_plain::<[u8; 4], SrcState>(bpf, "V4_SRC_STATE")?; + state.src_v6 = scan_plain::<[u8; 16], SrcState>(bpf, "V6_SRC_STATE")?; + state.src_sampled_ns = crate::gc::monotonic_now_ns(); // --- TX (egress) rates per local IP (display only; no mitigation) --- compute_tx::<[u8; 4]>( @@ -659,7 +539,6 @@ pub fn run_control( 32, &mut state.v4, &cfg.detection, - sba4, cfg.escalate_pass_pps, cfg.syn_proxy_pps, now_ns, @@ -676,7 +555,6 @@ pub fn run_control( mask_v4, &mut state.prefix_v4, &cfg.network, - sba4, cfg.escalate_pass_pps, cfg.syn_proxy_pps, now_ns, @@ -696,7 +574,6 @@ pub fn run_control( 128, &mut state.v6, &cfg.detection, - sba6, cfg.escalate_pass_pps, cfg.syn_proxy_pps, now_ns, @@ -713,7 +590,6 @@ pub fn run_control( mask_v6, &mut state.prefix_v6, &cfg.network, - sba6, cfg.escalate_pass_pps, cfg.syn_proxy_pps, now_ns, diff --git a/lnvps_fw/lnvps_fw_service/tests/api.rs b/lnvps_fw/lnvps_fw_service/tests/api.rs index 2e823113..5330c815 100644 --- a/lnvps_fw/lnvps_fw_service/tests/api.rs +++ b/lnvps_fw/lnvps_fw_service/tests/api.rs @@ -243,7 +243,6 @@ async fn limits_put_get_roundtrip_and_validation() { exit_pct: 60, cooldown_secs: 45, src_rate_pps: 20_000, - src_exit_pct: 40, src_cooldown_secs: 15, }) .unwrap(); @@ -265,7 +264,6 @@ async fn limits_put_get_roundtrip_and_validation() { let got: Limits = body_json(res).await; assert_eq!(got.exit_pct, 60); assert_eq!(got.src_rate_pps, 20_000); - assert_eq!(got.src_exit_pct, 40); assert_eq!(got.src_cooldown_secs, 15); // Zero threshold rejected. diff --git a/lnvps_fw/lnvps_fw_service/tests/carpet_bomb.rs b/lnvps_fw/lnvps_fw_service/tests/carpet_bomb.rs index 8fcb2342..afa0dc6a 100644 --- a/lnvps_fw/lnvps_fw_service/tests/carpet_bomb.rs +++ b/lnvps_fw/lnvps_fw_service/tests/carpet_bomb.rs @@ -46,15 +46,8 @@ fn thin_carpet_bomb_flips_whole_prefix() { manual_v4: Vec::new(), manual_v6: Vec::new(), src_rate_pps: u64::MAX, - fanout: 4, - agg_max_prefix_v4: 24, - agg_max_prefix_v6: 48, - src_exit_pct: 50, src_cooldown_ns: SECOND_NS, - max_source_blocks: 50_000, - block_ttl_ns: SECOND_NS, escalate_pass_pps: u64::MAX, - max_real_sources: 10_000, syn_proxy_pps: u64::MAX, }; let mut state = DetectionState::default(); diff --git a/lnvps_fw/lnvps_fw_service/tests/escalation.rs b/lnvps_fw/lnvps_fw_service/tests/escalation.rs index 0abea8ed..aa941ab6 100644 --- a/lnvps_fw/lnvps_fw_service/tests/escalation.rs +++ b/lnvps_fw/lnvps_fw_service/tests/escalation.rs @@ -1,118 +1,154 @@ -//! Increment-5 harness tests: per-source rate detection + CIDR aggregation. -//! -//! The eBPF side only counts per source; userspace computes per-source rates -//! and installs aggregated CIDR blocks. Root-only and `#[ignore]`d; run with -//! `scripts/fw-e2e.sh --test escalation`. +//! In-kernel per-source rate machine: the XDP datapath computes each source's +//! window rate and blocks over-rate sources on its own — **no userspace +//! control ticks are involved in the decision**. Userspace only arms the +//! config map and reads the state for display. Root-only and `#[ignore]`d; +//! run with `scripts/fw-e2e.sh --test escalation`. mod harness; use std::net::Ipv4Addr; +use std::thread::sleep; +use std::time::Duration; use harness::netns::VM_V4; use harness::traffic; use harness::{Harness, require_root}; -use lnvps_fw_service::detect::DetectionConfig; -use lnvps_fw_service::runtime::{DetectionState, RuntimeConfig}; +use lnvps_fw_common::{DEST_MODE_PORT_FILTER, DEST_MODE_SOURCE_BLOCK}; const SECOND_NS: u64 = 1_000_000_000; -/// Per-source state machine: offending sources are blocked as individual /32s -/// (NOT aggregated into a /24 while there is trie space), a low-rate neighbour -/// in the same /24 is never caught, and a blocked source is RELEASED once its -/// rate falls below the exit threshold for the cooldown — not held for a blind -/// TTL. Uses the real `runtime::run_control`. +/// Offending sources are blocked by the kernel itself: a source exceeding the +/// per-window limit toward a SOURCE_BLOCK-escalated destination trips +/// `blocked_until` in its state entry and its packets are dropped, while a +/// low-rate source (same or different /24) is never touched. Once the flood +/// stops, the block expires after the cooldown without any userspace action. #[test] #[ignore = "requires root / CAP_NET_ADMIN"] -fn per_source_blocks_slash32_and_releases_on_hysteresis() { +fn kernel_blocks_over_rate_source_and_releases_after_cooldown() { if !require_root() { return; } let mut h = Harness::new().expect("harness setup"); - h.set_mitigate_v4(VM_V4).expect("mitigate"); + // Escalated destination: port filter + source blocking enforced. + h.set_dest_flags_v4(VM_V4, 32, DEST_MODE_PORT_FILTER | DEST_MODE_SOURCE_BLOCK) + .expect("mitigate"); + // Arm the kernel rate machine: >10 packets within a 1s window blocks the + // source for a 1s cooldown. + h.set_src_rate(10, SECOND_NS).expect("src rate cfg"); - let quiet = DetectionConfig { - pps: u64::MAX, - syn_pps: u64::MAX, - bps: u64::MAX, - exit_pct: 50, - cooldown_ns: SECOND_NS, - }; - let cfg = RuntimeConfig { - // Keep dest + prefix detection out of the way; drive DEST_STATE manually. - detection: quiet, - network: quiet, - protected_v4: Vec::new(), - protected_v6: Vec::new(), - manual_v4: Vec::new(), - manual_v6: Vec::new(), - src_rate_pps: 10, // a source sending >=10pps trips DROPPING - fanout: 4, - agg_max_prefix_v4: 24, - agg_max_prefix_v6: 48, - src_exit_pct: 50, // exit below 5pps - src_cooldown_ns: SECOND_NS, - max_source_blocks: 50_000, // ample space -> /32s, no aggregation - block_ttl_ns: SECOND_NS, - escalate_pass_pps: 0, - max_real_sources: 10_000, - syn_proxy_pps: u64::MAX, - }; - let mut state = DetectionState::default(); + let offender = Ipv4Addr::new(10, 0, 9, 1); + let neighbour = Ipv4Addr::new(10, 0, 9, 200); // same /24, low rate + let unrelated = Ipv4Addr::new(10, 0, 99, 5); - // t0: seed snapshots (no traffic yet). - h.run_control_tick(&mut state, &cfg, SECOND_NS).expect("t0"); + // The offender bursts 50 packets (all within one window); the others send 3. + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[offender], VM_V4, 9999, 50).expect("flood"); + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[neighbour, unrelated], VM_V4, 9999, 3) + .expect("low"); - // Four sources in 10.0.9.0/24 flood (20 pkts ~= 20pps over the 1s window); - // one unrelated source sends only 5 (5pps, below the offender threshold). - let offenders: Vec = (1..=4).map(|i| Ipv4Addr::new(10, 0, 9, i)).collect(); - let unrelated = Ipv4Addr::new(10, 0, 99, 5); - traffic::udp_flood_sources_v4(&attacker_ns(&h), &offenders, VM_V4, 9999, 20).expect("flood"); - traffic::udp_flood_sources_v4(&attacker_ns(&h), &[unrelated], VM_V4, 9999, 5).expect("low"); + // The kernel counted and decided on its own — no control tick has run. + assert!( + h.src_packets_v4(offender).unwrap() >= 10, + "offender packets counted in the kernel window" + ); + assert!( + h.src_blocked_v4(offender).unwrap(), + "kernel blocked the over-rate source without userspace" + ); + assert!( + !h.src_blocked_v4(neighbour).unwrap(), + "low-rate neighbour in the same /24 stays unblocked" + ); + assert!( + !h.src_blocked_v4(unrelated).unwrap(), + "low-rate unrelated source stays unblocked" + ); + // Blocked means dropped: further offender packets must not reach the VM. + let before = h + .dest_counters_v4(VM_V4) + .unwrap() + .unwrap_or_default() + .dropped; + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[offender], VM_V4, 9999, 10) + .expect("blocked flood"); + let after = h + .dest_counters_v4(VM_V4) + .unwrap() + .unwrap_or_default() + .dropped; assert!( - h.src_packets_v4(Ipv4Addr::new(10, 0, 9, 1)).unwrap() >= 20, - "per-source packets should be counted" + after > before, + "packets from a kernel-blocked source are dropped ({before} -> {after})" ); - // t1: one control tick moves the offenders into DROPPING and blocks each /32. - h.run_control_tick(&mut state, &cfg, 2 * SECOND_NS) - .expect("t1"); + // Flood stops: the block expires by itself after the cooldown (the last + // over-rate packet extended it by 1s at most). + sleep(Duration::from_millis(2200)); + assert!( + !h.src_blocked_v4(offender).unwrap(), + "block expires after the cooldown once the flood stops" + ); +} - for o in &offenders { - assert!( - h.cidr_blocked_v4(*o).unwrap(), - "offending source {o} should be blocked as a /32" - ); +/// Enforcement is gated on the destination's SOURCE_BLOCK escalation: with +/// only PORT_FILTER set, an over-rate source is *counted* (and marked blocked +/// in its state) but its packets are not source-dropped — the escalation +/// ladder still decides when source blocking engages. +#[test] +#[ignore = "requires root / CAP_NET_ADMIN"] +fn source_drops_gated_on_source_block_flag() { + if !require_root() { + return; } - // Crucially: a non-offending neighbour in the SAME /24 must NOT be blocked - // (no eager aggregation while there is trie space). + let mut h = Harness::new().expect("harness setup"); + h.set_dest_flags_v4(VM_V4, 32, DEST_MODE_PORT_FILTER) + .expect("mitigate (no SOURCE_BLOCK)"); + h.set_src_rate(10, 60 * SECOND_NS).expect("src rate cfg"); + // Learn the target port as open so PORT_FILTER passes the traffic and any + // drop could only come from source blocking. + h.set_open_port_v4(VM_V4, 9999, lnvps_fw_common::PROTO_UDP) + .expect("open port"); + + let offender = Ipv4Addr::new(10, 0, 9, 1); + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[offender], VM_V4, 9999, 50).expect("flood"); + + // The kernel marked it blocked in its state map… assert!( - !h.cidr_blocked_v4(Ipv4Addr::new(10, 0, 9, 200)).unwrap(), - "a low-rate neighbour in the same /24 must not be caught by aggregation" + h.src_blocked_v4(offender).unwrap(), + "over-rate source is marked blocked in the state map" ); + // …but without the SOURCE_BLOCK flag its packets keep flowing. + let before = h + .dest_counters_v4(VM_V4) + .unwrap() + .unwrap_or_default() + .packets; + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[offender], VM_V4, 9999, 10).expect("more"); + let counters = h.dest_counters_v4(VM_V4).unwrap().unwrap_or_default(); assert!( - !h.cidr_blocked_v4(unrelated).unwrap(), - "low-rate unrelated source must not be blocked" + counters.packets >= before + 10, + "packets still counted/passed without SOURCE_BLOCK escalation" ); - // Flood stops. The source rate drops to 0; the state machine needs one tick - // to record "below exit" and then the cooldown (1s) to elapse before it - // returns to NORMAL and is unblocked. - h.run_control_tick(&mut state, &cfg, 3 * SECOND_NS) - .expect("cooldown-start"); + // Escalate: now the same source's packets are dropped. + h.set_dest_flags_v4(VM_V4, 32, DEST_MODE_PORT_FILTER | DEST_MODE_SOURCE_BLOCK) + .expect("escalate"); + let dropped_before = h + .dest_counters_v4(VM_V4) + .unwrap() + .unwrap_or_default() + .dropped; + traffic::udp_flood_sources_v4(&attacker_ns(&h), &[offender], VM_V4, 9999, 10) + .expect("blocked flood"); + let dropped_after = h + .dest_counters_v4(VM_V4) + .unwrap() + .unwrap_or_default() + .dropped; assert!( - h.cidr_blocked_v4(Ipv4Addr::new(10, 0, 9, 1)).unwrap(), - "still blocked during the cooldown window" + dropped_after > dropped_before, + "SOURCE_BLOCK escalation engages the kernel's block" ); - // A second tick a full cooldown later releases the block. - h.run_control_tick(&mut state, &cfg, 5 * SECOND_NS) - .expect("release"); - for o in &offenders { - assert!( - !h.cidr_blocked_v4(*o).unwrap(), - "source {o} released once its rate fell below exit for the cooldown" - ); - } } fn attacker_ns(h: &Harness) -> String { diff --git a/lnvps_fw/lnvps_fw_service/tests/harness/mod.rs b/lnvps_fw/lnvps_fw_service/tests/harness/mod.rs index efc8e24e..a9c6e3c5 100644 --- a/lnvps_fw/lnvps_fw_service/tests/harness/mod.rs +++ b/lnvps_fw/lnvps_fw_service/tests/harness/mod.rs @@ -29,10 +29,36 @@ use nix::sched::{CloneFlags, setns}; use lnvps_fw_common::{ DEST_MODE_PORT_FILTER, DestCounters, DestState, LastSeen, PortKeyV4, PortKeyV6, - SLOT_SYN_PROXY_V4, SLOT_SYN_PROXY_V6, + SLOT_SYN_PROXY_V4, SLOT_SYN_PROXY_V6, SrcState, }; +use lnvps_fw_service::detect::DetectionConfig; use lnvps_fw_service::runtime::{DetectionState, RuntimeConfig, run_control}; +/// A `RuntimeConfig` with every dest/prefix detector quiet (thresholds at +/// `u64::MAX`) so harness tests can drive DEST_STATE and the per-source +/// machine explicitly without auto-detection interfering. +pub fn test_runtime_config() -> RuntimeConfig { + let quiet = DetectionConfig { + pps: u64::MAX, + syn_pps: u64::MAX, + bps: u64::MAX, + exit_pct: 50, + cooldown_ns: 1_000_000_000, + }; + RuntimeConfig { + detection: quiet, + network: quiet, + protected_v4: Vec::new(), + protected_v6: Vec::new(), + manual_v4: Vec::new(), + manual_v6: Vec::new(), + src_rate_pps: u64::MAX, + src_cooldown_ns: 1_000_000_000, + escalate_pass_pps: 0, + syn_proxy_pps: u64::MAX, + } +} + use netns::NetnsTopology; /// The compiled eBPF object, produced by the package build script @@ -168,15 +194,13 @@ impl Harness { /// the shared `gc` logic so the harness test exercises real code. pub fn gc_open_ports_v4(&mut self, ttl_ns: u64) -> anyhow::Result { let now = lnvps_fw_service::gc::monotonic_now_ns(); + let entries: Vec<(PortKeyV4, LastSeen)> = + lnvps_fw_service::runtime::scan_plain(&self.bpf, "OPEN_PORTS_V4")?; + let expired = + lnvps_fw_service::gc::expired_ports(&entries, now, ttl_ns, ttl_ns, |k| k.proto); let mut map: AyaHashMap<_, PortKeyV4, LastSeen> = AyaHashMap::try_from(self.bpf.map_mut("OPEN_PORTS_V4").unwrap())?; - Ok(lnvps_fw_service::gc::gc_open_ports( - &mut map, - now, - ttl_ns, - ttl_ns, - |k| k.proto, - )) + Ok(lnvps_fw_service::gc::remove_keys(&mut map, &expired)) } /// Force an IPv4 destination (or prefix) into MITIGATE mode by writing the @@ -224,14 +248,6 @@ impl Harness { Ok(()) } - /// Manually install a source CIDR block (as escalation would). - pub fn block_cidr_v4(&mut self, net: Ipv4Addr, prefix_len: u32) -> anyhow::Result<()> { - let mut trie: LpmTrie<_, [u8; 4], u8> = - LpmTrie::try_from(self.bpf.map_mut("V4_CIDR_SRC").unwrap())?; - trie.insert(&Key::new(prefix_len, net.octets()), 1u8, 0)?; - Ok(()) - } - /// Read the effective mitigation mode covering an IPv4 destination via a /// longest-prefix lookup (default NORMAL = 0 if no covering entry). pub fn dest_mode_v4(&self, ip: Ipv4Addr) -> anyhow::Result { @@ -255,21 +271,36 @@ impl Harness { run_control(&mut self.bpf, state, cfg, now_ns) } - /// True if `ip` is covered by a blocked source CIDR in `V4_CIDR_SRC`. - pub fn cidr_blocked_v4(&self, ip: Ipv4Addr) -> anyhow::Result { - let trie: LpmTrie<_, [u8; 4], u8> = - LpmTrie::try_from(self.bpf.map("V4_CIDR_SRC").unwrap())?; - Ok(trie.get(&Key::new(32, ip.octets()), 0).is_ok()) + /// The kernel rate machine's state for a source (None if never counted). + pub fn src_state_v4(&self, ip: Ipv4Addr) -> anyhow::Result> { + let map: AyaHashMap<_, [u8; 4], SrcState> = + AyaHashMap::try_from(self.bpf.map("V4_SRC_STATE").unwrap())?; + Ok(map.get(&ip.octets(), 0).ok()) } - /// Summed per-CPU per-source packet count for `ip` (0 if absent). + /// True while the kernel machine is blocking `ip` (`blocked_until` in the + /// future of the shared CLOCK_MONOTONIC / bpf_ktime domain). + pub fn src_blocked_v4(&self, ip: Ipv4Addr) -> anyhow::Result { + let now = lnvps_fw_service::gc::monotonic_now_ns(); + Ok(self + .src_state_v4(ip)? + .is_some_and(|st| st.blocked_until_ns > now)) + } + + /// Packets the kernel counted for `ip` in its current window. pub fn src_packets_v4(&self, ip: Ipv4Addr) -> anyhow::Result { - let map: PerCpuHashMap<_, [u8; 4], u64> = - PerCpuHashMap::try_from(self.bpf.map("V4_SRC_COUNTERS").unwrap())?; - match map.get(&ip.octets(), 0) { - Ok(values) => Ok(values.iter().copied().sum()), - Err(_) => Ok(0), - } + Ok(self.src_state_v4(ip)?.map(|st| st.count).unwrap_or(0)) + } + + /// Arm the in-kernel per-source rate machine (as the daemon does at + /// startup and on limit edits). + pub fn set_src_rate(&mut self, rate_pps: u64, cooldown_ns: u64) -> anyhow::Result<()> { + let cfg = RuntimeConfig { + src_rate_pps: rate_pps, + src_cooldown_ns: cooldown_ns, + ..test_runtime_config() + }; + lnvps_fw_service::runtime::write_src_rate_cfg(&mut self.bpf, &cfg) } /// Seed a learned-open IPv4 port directly (as passive learning would). diff --git a/lnvps_fw/lnvps_fw_service/tests/mitigation.rs b/lnvps_fw/lnvps_fw_service/tests/mitigation.rs index 890a294e..ab0dea2f 100644 --- a/lnvps_fw/lnvps_fw_service/tests/mitigation.rs +++ b/lnvps_fw/lnvps_fw_service/tests/mitigation.rs @@ -7,13 +7,10 @@ mod harness; use std::net::SocketAddr; use std::time::Duration; -use harness::netns::{ATTACKER_V4, VM_V4}; +use harness::netns::VM_V4; use harness::traffic; use harness::{Harness, require_root}; -use lnvps_fw_common::{ - DEST_MODE_NORMAL, DEST_MODE_PORT_FILTER, DEST_MODE_SOURCE_BLOCK, DEST_MODE_SYN_PROXY, - PROTO_TCP, PROTO_UDP, -}; +use lnvps_fw_common::{DEST_MODE_NORMAL, DEST_MODE_PORT_FILTER, DEST_MODE_SYN_PROXY, PROTO_TCP}; use lnvps_fw_service::detect::DetectionConfig; use lnvps_fw_service::runtime::{DetectionState, RuntimeConfig}; @@ -123,15 +120,8 @@ fn detection_flip_and_cooldown() { manual_v4: Vec::new(), manual_v6: Vec::new(), src_rate_pps: u64::MAX, // don't block sources in this test - fanout: 4, - agg_max_prefix_v4: 24, - agg_max_prefix_v6: 48, - src_exit_pct: 50, src_cooldown_ns: SECOND_NS, - max_source_blocks: 50_000, - block_ttl_ns: SECOND_NS, escalate_pass_pps: u64::MAX, - max_real_sources: 10_000, syn_proxy_pps: u64::MAX, }; let mut state = DetectionState::default(); @@ -204,15 +194,8 @@ fn manual_override_survives_auto_detection() { manual_v4: vec![(32, VM_V4.octets(), DEST_MODE_SYN_PROXY)], manual_v6: Vec::new(), src_rate_pps: u64::MAX, - fanout: 4, - agg_max_prefix_v4: 24, - agg_max_prefix_v6: 48, - src_exit_pct: 50, src_cooldown_ns: SECOND_NS, - max_source_blocks: 50_000, - block_ttl_ns: SECOND_NS, escalate_pass_pps: u64::MAX, - max_real_sources: 10_000, syn_proxy_pps: u64::MAX, }; let mut state = DetectionState::default(); @@ -248,61 +231,6 @@ fn manual_override_survives_auto_detection() { ); } -/// A CIDR-blocked source is only dropped when the SOURCE_BLOCK flag is set, not -/// with PORT_FILTER alone — protection flags are independent and source blocking -/// is only enabled when userspace decides it's warranted. -#[test] -#[ignore = "requires root / CAP_NET_ADMIN"] -fn source_block_only_when_flag_set() { - if !require_root() { - return; - } - let mut h = Harness::new().expect("harness setup"); - - // Learn an open UDP port so the port filter would PASS this traffic; then - // any drop we observe is attributable to source blocking, not the port gate. - traffic::udp_send_from( - &vm_ns(&h), - 7000, - SocketAddr::from((ATTACKER_V4, 9999)), - b"learn", - ) - .expect("learn port"); - std::thread::sleep(Duration::from_millis(200)); - assert!(h.open_port_v4(VM_V4, 7000, PROTO_UDP).unwrap().is_some()); - - // Block the attacker's source and mitigate the dest with only PORT_FILTER. - h.block_cidr_v4(ATTACKER_V4, 32).expect("block cidr"); - h.set_dest_flags_v4(VM_V4, 32, DEST_MODE_PORT_FILTER) - .expect("port filter only"); - - let open = SocketAddr::from((VM_V4, 7000)); - let before = h - .dest_counters_v4(VM_V4) - .unwrap() - .map(|c| c.dropped) - .unwrap_or(0); - traffic::udp_send_burst(&attacker_ns(&h), open, 10).expect("send lvl1"); - let mid = h.dest_counters_v4(VM_V4).unwrap().unwrap().dropped; - assert_eq!( - mid - before, - 0, - "at PORT_FILTER a blocked source to an open port must pass" - ); - - // Add the SOURCE_BLOCK flag: now the blocked source is dropped even though - // the port filter would have passed it (flags are independent). - h.set_dest_flags_v4(VM_V4, 32, DEST_MODE_PORT_FILTER | DEST_MODE_SOURCE_BLOCK) - .expect("port filter + source block"); - traffic::udp_send_burst(&attacker_ns(&h), open, 10).expect("send lvl4"); - let after = h.dest_counters_v4(VM_V4).unwrap().unwrap().dropped; - assert!( - after - mid >= 10, - "at SOURCE_BLOCK the blocked source must be dropped (delta={})", - after - mid - ); -} - fn attacker_ns(h: &Harness) -> String { format!("/var/run/netns/{}", h.topo.attacker_ns) } diff --git a/work/fw-inkernel-src-rate.md b/work/fw-inkernel-src-rate.md new file mode 100644 index 00000000..b5f10f9e --- /dev/null +++ b/work/fw-inkernel-src-rate.md @@ -0,0 +1,98 @@ +# In-kernel per-source rate limiting (XDP rewrite) + +**Status:** complete +**Started:** 2026-07-11 +**Last updated:** 2026-07-11 + +## Goal + +Move the per-source pps calculation and blocking decision from the userspace +control loop into the XDP datapath. Userspace stops scanning the (up to 256k +entry) source counter maps every 500ms tick and only reads state for display. +The class of bugs this kills structurally: cumulative-counter seeding, hidden +threshold drift between the two rate systems, and control-loop CPU that scales +with flood history instead of live traffic. + +## Design + +- `V4/V6_SRC_COUNTERS` (per-CPU cumulative u64) are **replaced** by + `V4/V6_SRC_STATE: LruHashMap` where + `SrcState { window_start_ns, count, blocked_until_ns }`. +- Fixed-window rate machine, in-kernel, per packet under mitigation: + - blocked? (`now < blocked_until_ns`) → count, and on window roll while + still over-rate extend the block (re-trip); XDP_DROP. + - window rolled (`now - window_start >= window_ns`) → reset window. + - `count++` (atomic); `count > max_per_window` → `blocked_until = now + + cooldown_ns`, XDP_DROP. +- Config via `SRC_RATE_CFG: Array` (1 entry): + `{ max_per_window, window_ns, cooldown_ns, enforce }` — written by userspace + at startup and on `PUT /limits`. `max_per_window` is precomputed + (`rate_pps × window_secs`) so the datapath never divides. +- Counting happens under any mitigation flags (same as today); **dropping** is + gated on the dest's `SOURCE_BLOCK` flag (escalation ladder unchanged: + userspace still decides *when* a dest escalates via `escalate_pass_pps`). +- Auto CIDR aggregation, spoof gate (`max_real_sources`), `plan_blocks`, + `V4/V6_CIDR_SRC` auto entries: **deleted**. The per-source state map IS the + block list; a spoofed flood of unique IPs never trips per-source limits and + simply churns the LRU (port-filter layer remains the defense, as today). + Manual blocks keep the separate `MANUAL_BLOCK_V4/V6` tries (unchanged). +- Userspace `/sources` + `/blocks` views: batched on-demand read of the state + maps (still via `batch.rs`), `state = blocked_until > now ? dropping : + normal` (the `cooling` display state disappears), `pps` approximated from + the current window. `src_exit_pct` is retired from `Limits`; + `src_rate_pps`/`src_cooldown_secs` remain and now write `SRC_RATE_CFG`. + +## Findings + +- Prior perf work (uncommitted, kept): `batch.rs` raw `BPF_MAP_LOOKUP_BATCH` + reader + batched dest/tx/GC scans + `block_pps` de-quadratic + idle purge. + The purge/tracker machinery (`detect.rs` SourceTracker, `step_sources`) gets + deleted by increment 2 of this rewrite. +- `mitigate_v4` at `lnvps_ebpf/src/main.rs:260` (`count_src_v4` + + `cidr_blocked_v4`) and v6 at `:315` are the integration points. +- Harness tests that break: `tests/escalation.rs` + (`cidr_escalation_blocks_offending_v24` — tests deleted aggregation), + `tests/mitigation.rs` (`source_block_only_when_flag_set` — same semantics, + new map). Harness accessors live in `tests/harness/mod.rs`. +- eBPF atomics: `count` increments use BPF atomic add; per-entry races on + window reset are benign (approximate counting is acceptable). + +## Tasks + +- [x] Pre-work: batch reader + GC batching compiles green (63 lib tests) +- [x] Increment 1: `SrcState`/`SrcRateConfig` in `lnvps_fw_common`; eBPF + `V4/V6_SRC_STATE` maps + `SRC_RATE_CFG`; rate machine in + `mitigate_v4/v6`; deleted `V4/V6_SRC_COUNTERS` + `V4/V6_CIDR_SRC` +- [x] Increment 2: stripped userspace source polling (SourceTracker, + step_sources, plan_blocks/aggregation, spoof gate, block trie + reconciliation); `write_src_rate_cfg` at startup + on `PUT /limits`; + `gc_src_states` on the GC timer (60s idle TTL); views from batched + state-map snapshots; deprecated config knobs kept parseable (regression + test `parses_legacy_escalation_keys`); `src_exit_pct` removed from + `Limits` +- [x] Increment 3: harness accessors rewritten (`src_state_v4`, + `src_blocked_v4`, `set_src_rate`); `escalation.rs` rewritten as two + in-kernel scenarios (kernel blocks over-rate source + releases after + cooldown with **zero** control ticks; drops gated on SOURCE_BLOCK); + superseded `source_block_only_when_flag_set` removed; full e2e suite + green (21 root tests: smoke 4, learning 3, mitigation 4, escalation 2, + carpet_bomb 1, syn_proxy 3, gre_decap 2, scoping 2); docs + example + config + dashboard updated, dist rebuilt + +## Outcome + +43 lib + 2 bin + 15 API tests and the full netns e2e suite pass. The control +loop's per-tick source work went from "iterate every counted source × 2 +syscalls" to one batched display read; blocking latency went from up to one +500ms tick to the packet that crosses the limit; seeding/counter-lifecycle +bugs are structurally gone. Deployment note: on upgrade the old +`V4/V6_CIDR_SRC`-based auto blocks disappear (kernel re-blocks offenders +within one window); old config files parse unchanged. + +## Notes + +- Do NOT release mid-rewrite; single release once increment 3 is green. +- Keep `GET /blocks` API shape backward-compatible (manual + dropping /32s). +- `escalate_pass_pps` and the dest-level detection ladder are intentionally + untouched — dest detection stays in userspace (16k entries, cheap, complex + policy).