test(flow-filter): carry the config oracle through to the NF, and widen the generators

config/src/external/overlay/vpc.rs

@@ -151,8 +151,7 @@ impl ValidatedPeering {
     }
 
     fn validate_nat_combinations(&self) -> ConfigResult {
-        // If stateful NAT is set up on one side of the peering, we don't support NAT (static or
-        // stateful) on the other side.
+        // Stateful NAT cannot appear on both sides; static NAT is compatible with every mode.
         let mut local_has_masquerading = false;
         let mut local_has_port_forwarding = false;
         for expose in self.local.valexp() {

flow-filter/Cargo.toml

@@ -32,4 +32,5 @@ acl = { workspace = true, features = ["reference"] }
 bolero = { workspace = true, features = ["std"] }
 dpdk = { workspace = true, features = ["test"] }
 lpm = { workspace = true, features = ["testing"] }
-net = { workspace = true, features = ["builder"] }
+# Enable generated header stacks for classifier tests.
+net = { workspace = true, features = ["builder", "bolero"] }

flow-filter/src/context/fuzz.rs

@@ -67,7 +67,8 @@ fn consider<T>(best: &mut Option<(Precedence, T)>, precedence: Precedence, value
 }
 
 /// Answer a route lookup directly from the validated overlay.
-fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult {
+/// Shared by the context and NF metadata property tests.
+pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult {
     let Some(src_vpc) = overlay
         .vpc_table()
         .values()
@@ -364,3 +365,43 @@ fn reference_lookup_matches_config_oracle() {
             }
         });
 }
+
+/// Require generated exclusions to produce multi-length prefix fans.
+/// Removing one host from a `/24` must produce lengths `/25` through `/32`.
+#[test]
+fn exclusions_reach_the_config_as_multi_length_prefix_fans() {
+    use std::collections::BTreeSet;
+
+    // Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not const.
+    static WIDEST_SPREAD: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
+
+    bolero::check!()
+        .with_type::<OverlaySpec>()
+        .for_each(|overlay_spec| {
+            let built = overlay_spec.build();
+            for vpc in built.overlay.vpc_table().values() {
+                for peering in vpc.peerings() {
+                    let exposes = peering
+                        .local()
+                        .valexp()
+                        .iter()
+                        .chain(peering.remote().valexp());
+                    for expose in exposes {
+                        for set in [expose.ips(), expose.public_ips()] {
+                            let lengths: BTreeSet<u8> =
+                                set.iter().map(|p| p.prefix().length()).collect();
+                            WIDEST_SPREAD.fetch_max(lengths.len() as u64, Ordering::Relaxed);
+                        }
+                    }
+                }
+            }
+        });
+
+    let spread = WIDEST_SPREAD.load(Ordering::Relaxed);
+    eprintln!("coverage: widest prefix-length spread in a single expose: {spread}");
+    assert!(
+        spread >= 8,
+        "exclusions never produced a full prefix-length fan (widest spread was {spread}); \
+         the generator is emitting single-block exposes and the priority ordering is untested",
+    );
+}

flow-filter/src/context/mod.rs

@@ -10,7 +10,7 @@ use config::external::overlay::ValidatedOverlay;
 
 mod display;
 #[cfg(test)]
-mod fuzz;
+pub(crate) mod fuzz;
 mod tables;
 #[cfg(test)]
 mod tests;

flow-filter/src/fuzz_gen.rs

@@ -71,6 +71,14 @@ impl FwProto {
             FwProto::Udp => Some(L4Protocol::Udp),
         }
     }
+
+    /// A protocol accepted by this expose. Generated `Any` probes use TCP.
+    fn probe_next_header(self) -> NextHeader {
+        match self {
+            FwProto::Any | FwProto::Tcp => NextHeader::TCP,
+            FwProto::Udp => NextHeader::UDP,
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, TypeGenerator)]
@@ -95,10 +103,6 @@ impl ExposeSpec {
         !matches!(self, ExposeSpec::Plain | ExposeSpec::StaticNat)
     }
 
-    fn has_nat(self) -> bool {
-        !matches!(self, ExposeSpec::Plain)
-    }
-
     /// Whether this expose gives the source side of a route an unconstrained, connection-initiating
     /// match: a plain / static-nat / masquerade private block (a `/24` or `/120` with no port
     /// constraint, `can_init_connection`). Port forwarding cannot initiate, so a pure
@@ -110,10 +114,19 @@ impl ExposeSpec {
         )
     }
 
-    /// Where a destination address this expose matches lives, or `None` if it only matches
-    /// port-forwarded destinations (skipped -- those need a specific public port). `Some(true)`
-    /// means the public block (NAT exposes translate destinations into it); `Some(false)` means the
-    /// private block (a plain expose's public IPs are its private IPs).
+    /// Protocols for probes targeting this expose's port-forwarded destination.
+    fn portfw_protos(self) -> Vec<FwProto> {
+        match self {
+            ExposeSpec::Plain | ExposeSpec::StaticNat | ExposeSpec::Masquerade => Vec::new(),
+            ExposeSpec::PortForwarding(proto)
+            | ExposeSpec::MasqueradeNestingPortFw(proto)
+            | ExposeSpec::MasqueradeSameLenPortFw(proto) => vec![proto],
+            ExposeSpec::PortFwProtoPair => vec![FwProto::Tcp, FwProto::Udp],
+        }
+    }
+
+    /// Whether a matching destination uses the public or private block.
+    /// Returns `None` for exposes that require a port-forwarding probe.
     fn dest_public_space(self) -> Option<bool> {
         match self {
             ExposeSpec::Plain => Some(false),
@@ -126,19 +139,74 @@ impl ExposeSpec {
     }
 }
 
+/// An exclusion applied to an expose's private and public blocks.
+/// All selections preserve probe host `.1` and port-forwarding host [`FW_HOST`].
+#[derive(Debug, Clone, Copy, TypeGenerator)]
+pub(crate) enum ExcludeSel {
+    None,
+    /// The block's upper half (a `/25`, or `/121` for v6).
+    UpperHalf,
+    /// The block's second quarter, leaving runs on both sides of the hole.
+    SecondQuarter,
+    /// A single host in the upper half: the widest fan of prefix lengths a single exclusion can
+    /// produce.
+    UpperHost(u8),
+}
+
+impl ExcludeSel {
+    /// The prefix to exclude from block `n`, or `None` to leave the block whole.
+    fn resolve(self, n: u8, public: bool, v6: bool) -> Option<String> {
+        let (net4, net6) = if public { (20, "db9") } else { (10, "db8") };
+        Some(match (self, v6) {
+            (ExcludeSel::None, _) => return None,
+            (ExcludeSel::UpperHalf, false) => format!("{net4}.{n}.0.128/25"),
+            (ExcludeSel::UpperHalf, true) => format!("2001:{net6}:0:{n:x}::80/121"),
+            (ExcludeSel::SecondQuarter, false) => format!("{net4}.{n}.0.64/26"),
+            (ExcludeSel::SecondQuarter, true) => format!("2001:{net6}:0:{n:x}::40/122"),
+            // Keep probe and port-forwarding hosts below the exclusion.
+            (ExcludeSel::UpperHost(h), false) => {
+                format!("{net4}.{n}.0.{}/32", h | 0x80)
+            }
+            (ExcludeSel::UpperHost(h), true) => {
+                format!("2001:{net6}:0:{n:x}::{:x}/128", h | 0x80)
+            }
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy, TypeGenerator)]
+pub(crate) struct ExposeEntry {
+    spec: ExposeSpec,
+    /// Exclusion for this expose. Port forwarding ignores it because config forbids exclusions.
+    exclude: ExcludeSel,
+}
+
+impl ExposeEntry {
+    const fn plain() -> Self {
+        Self {
+            spec: ExposeSpec::Plain,
+            exclude: ExcludeSel::None,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, TypeGenerator)]
 pub(crate) struct ManifestSpec {
     /// Up to two expose specs (some expand to two actual exposes).
-    exposes: [Option<ExposeSpec>; 2],
+    exposes: [Option<ExposeEntry>; 2],
     /// Whether the manifest carries a default (catch-all) expose.
     default: bool,
 }
 
 impl ManifestSpec {
-    fn expose_specs(&self) -> impl Iterator<Item = ExposeSpec> + '_ {
+    fn entries(&self) -> impl Iterator<Item = ExposeEntry> + '_ {
         self.exposes.iter().flatten().copied()
     }
 
+    fn expose_specs(&self) -> impl Iterator<Item = ExposeSpec> + '_ {
+        self.entries().map(|entry| entry.spec)
+    }
+
     fn is_empty(&self) -> bool {
         self.exposes.iter().all(Option::is_none) && !self.default
     }
@@ -147,22 +215,21 @@ impl ManifestSpec {
         self.expose_specs().any(ExposeSpec::is_stateful)
     }
 
-    fn has_nat(&self) -> bool {
-        self.expose_specs().any(ExposeSpec::has_nat)
-    }
-
-    fn strip_nat(&mut self) {
+    /// Replace every stateful-NAT expose with a static-NAT one.
+    ///
+    /// This makes both peering sides compatible while retaining static-NAT combinations.
+    fn strip_stateful_nat(&mut self) {
         for slot in self.exposes.iter_mut().flatten() {
-            if slot.has_nat() {
-                *slot = ExposeSpec::Plain;
+            if slot.spec.is_stateful() {
+                slot.spec = ExposeSpec::StaticNat;
             }
         }
     }
 
     fn drop_default(&mut self) {
         self.default = false;
         if self.is_empty() {
-            self.exposes[0] = Some(ExposeSpec::Plain);
+            self.exposes[0] = Some(ExposeEntry::plain());
         }
     }
 }
@@ -199,30 +266,28 @@ impl OverlaySpec {
             spec.peerings[0] = Some(PeeringSpec {
                 v6: false,
                 local: ManifestSpec {
-                    exposes: [Some(ExposeSpec::Plain), None],
+                    exposes: [Some(ExposeEntry::plain()), None],
                     default: false,
                 },
                 remote: ManifestSpec {
-                    exposes: [Some(ExposeSpec::Plain), None],
+                    exposes: [Some(ExposeEntry::plain()), None],
                     default: false,
                 },
             });
         }
         for peering in spec.peerings.iter_mut().flatten() {
             for manifest in [&mut peering.local, &mut peering.remote] {
                 if manifest.is_empty() {
-                    manifest.exposes[0] = Some(ExposeSpec::Plain);
+                    manifest.exposes[0] = Some(ExposeEntry::plain());
                 }
             }
             // A default expose cannot face another default expose within one peering.
             if peering.local.default && peering.remote.default {
                 peering.remote.drop_default();
             }
-            // Stateful NAT on one side of a peering forbids any NAT on the other side.
-            if peering.local.has_stateful() && peering.remote.has_nat() {
-                peering.remote.strip_nat();
-            } else if peering.remote.has_stateful() && peering.local.has_nat() {
-                peering.local.strip_nat();
+            // At most one side of a peering may use stateful NAT.
+            if peering.local.has_stateful() && peering.remote.has_stateful() {
+                peering.remote.strip_stateful_nat();
             }
         }
         // Each VPC may see at most one default destination across all of its peerings. A default
@@ -296,9 +361,9 @@ impl OverlaySpec {
     }
 }
 
-/// Append one guaranteed-routing probe for each (source-capable local expose, matchable remote
-/// expose) pair of a peering. The source lands at host `.1` of a can-init private block and the
-/// destination at host `.1` of the peer's matching block
+/// Append a routing probe for each compatible pair of local and remote exposes.
+///
+/// Port-forwarding probes target [`FW_HOST`] and [`FW_PUBLIC_PORTS`]. Other probes use host `.1`.
 fn derive_routing_probes(
     out: &mut Vec<Probe>,
     src_vni: u32,
@@ -313,16 +378,26 @@ fn derive_routing_probes(
         }
         let src_ip = block_addr(local_base + li as u8, 1, false, v6);
         for (ri, rspec) in remote.expose_specs().enumerate() {
-            let Some(dst_public) = rspec.dest_public_space() else {
-                continue;
-            };
-            out.push(Probe {
-                src_vpcd,
-                src_ip,
-                dst_ip: block_addr(remote_base + ri as u8, 1, dst_public, v6),
-                proto: NextHeader::TCP,
-                ports: Some((1, 1)),
-            });
+            let dst_block = remote_base + ri as u8;
+            if let Some(dst_public) = rspec.dest_public_space() {
+                out.push(Probe {
+                    src_vpcd,
+                    src_ip,
+                    dst_ip: block_addr(dst_block, 1, dst_public, v6),
+                    proto: NextHeader::TCP,
+                    ports: Some((1, 1)),
+                });
+            }
+            for proto in rspec.portfw_protos() {
+                out.push(Probe {
+                    src_vpcd,
+                    src_ip,
+                    dst_ip: block_addr(dst_block, FW_HOST, true, v6),
+                    proto: proto.probe_next_header(),
+                    // Source exposes do not constrain ports.
+                    ports: Some((1, FW_PUBLIC_PORTS.0)),
+                });
+            }
         }
     }
 }
@@ -333,19 +408,25 @@ fn vpc_name(index: usize) -> String {
 
 fn build_manifest(vpc_name: &str, spec: &ManifestSpec, v6: bool, blocks: &mut u8) -> VpcManifest {
     let mut exposes = Vec::new();
-    for expose_spec in spec.expose_specs() {
+    for entry in spec.entries() {
         let n = *blocks;
         *blocks += 1;
-        match expose_spec {
-            ExposeSpec::Plain => exposes.push(plain(n, v6)),
-            ExposeSpec::StaticNat => exposes.push(static_nat(n, v6)),
-            ExposeSpec::Masquerade => exposes.push(masquerade(n, v6)),
+        let exclude = entry.exclude;
+        match entry.spec {
+            ExposeSpec::Plain => exposes.push(excluding(plain(n, v6), n, v6, exclude)),
+            ExposeSpec::StaticNat => {
+                exposes.push(excluding_both(static_nat(n, v6), n, v6, exclude));
+            }
+            ExposeSpec::Masquerade => {
+                exposes.push(excluding_both(masquerade(n, v6), n, v6, exclude));
+            }
+            // Only the masquerade half may carry an exclusion.
             ExposeSpec::MasqueradeNestingPortFw(proto) => {
-                exposes.push(masquerade(n, v6));
+                exposes.push(excluding_both(masquerade(n, v6), n, v6, exclude));
                 exposes.push(portfw_host(n, v6, proto));
             }
             ExposeSpec::MasqueradeSameLenPortFw(proto) => {
-                exposes.push(masquerade(n, v6));
+                exposes.push(excluding_both(masquerade(n, v6), n, v6, exclude));
                 exposes.push(portfw_block(n, v6, proto));
             }
             ExposeSpec::PortForwarding(proto) => exposes.push(portfw_host(n, v6, proto)),
@@ -402,6 +483,28 @@ pub(crate) fn block_addr(n: u8, host: u8, public: bool, v6: bool) -> IpAddr {
     }
 }
 
+/// Remove `exclude` from an expose's private prefixes.
+fn excluding(expose: VpcExpose, n: u8, v6: bool, exclude: ExcludeSel) -> VpcExpose {
+    match exclude.resolve(n, false, v6) {
+        Some(prefix) => expose.not(prefix.as_str().into()),
+        None => expose,
+    }
+}
+
+/// Remove matching exclusions from both sides to preserve static NAT address counts.
+fn excluding_both(expose: VpcExpose, n: u8, v6: bool, exclude: ExcludeSel) -> VpcExpose {
+    let Some(private) = exclude.resolve(n, false, v6) else {
+        return expose;
+    };
+    let Some(public) = exclude.resolve(n, true, v6) else {
+        unreachable!("resolve is None only for ExcludeSel::None, handled above");
+    };
+    expose
+        .not(private.as_str().into())
+        .not_as(public.as_str().into())
+        .unwrap_or_else(|e| unreachable!("exclusion on an expose with a public range: {e}"))
+}
+
 fn plain(n: u8, v6: bool) -> VpcExpose {
     VpcExpose::empty().ip(private_block(n, v6).as_str().into())
 }

flow-filter/src/lib.rs

@@ -350,8 +350,8 @@ impl FlowFilter {
             return None;
         };
 
-        // The flow has the same generation id as the current config. Small transient period aside,
-        // this means that the flow is up-to-date and we can bypass the filter
+        // Current and newer-generation flows bypass the filter. Workers may observe a new config
+        // generation after flows have already been stamped with it.
         debug!("{nfi}: Packet can bypass flow filter thanks to flow information");
         Some(dst_vpcd)
     }