From 97baacd5036839b10189fd97206069bd05e4a973 Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 01:31:27 +0100 Subject: [PATCH 1/8] feat(flow-filter): Key masquerade destinations for flow re-validation Reply traffic for masqueraded flows don't usually require a flow-table lookup, because the associated flow table entry allows us to bypass the lookup under normal conditions. However, if the flow has an old genid (this may happen if we bump the genid for the flow-filter pipeline stage between the flow lookup and the flow-filter processing for the packet, for example), then we need to do the lookup to re-validate the flow. The issue with this lookup is that we may have colliding entries, if multiple VPCs expose (via masquerade) overlapping prefixes to a given VPC. To avoid that, we add the destination VNI as part of the key in the remote lookup table. The destination VNI is only used for reply traffic for masqueraded flows; for all other entries, it remains at 0 (invalid VNI number). On finding a packet with outdated flow-info for masquerade, we may run two consecutive lookups: one using the destination VNI in the key, and if it fails, a second one without it, for forward traffic (we cannot use the destination VNI for forward traffic entries, or we'd be unable to validate packets initiating new flows and without flow information). Signed-off-by: Quentin Monnet --- Cargo.lock | 1 + flow-filter/Cargo.toml | 1 + flow-filter/src/context/fuzz.rs | 9 +++ flow-filter/src/context/tables.rs | 95 ++++++++++++++++++++++++++++--- flow-filter/src/context/tests.rs | 28 +++++++-- flow-filter/src/lib.rs | 45 ++++++--------- 6 files changed, 137 insertions(+), 42 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 59f514d47e..c818995741 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1454,6 +1454,7 @@ dependencies = [ "indenter", "linkme", "tracing", + "tracing-test", ] [[package]] diff --git a/flow-filter/Cargo.toml b/flow-filter/Cargo.toml index 627c93494c..8ebae286b9 100644 --- a/flow-filter/Cargo.toml +++ b/flow-filter/Cargo.toml @@ -34,3 +34,4 @@ dpdk = { workspace = true, features = ["test"] } lpm = { workspace = true, features = ["testing"] } # Enable generated header stacks for classifier tests. net = { workspace = true, features = ["builder", "bolero"] } +tracing-test = { workspace = true } diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index 27eee723a2..31dcd771fb 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -178,6 +178,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { for probe in &built.routing_probes { let want = reference.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, @@ -186,6 +187,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { assert_eq!( dpdk.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, @@ -209,6 +211,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { .iter() .map(|p| LookupInput { src_vpcd: p.src_vpcd, + dst_vpcd: None, src_ip: p.src_ip, dst_ip: p.dst_ip, proto: p.proto, @@ -220,6 +223,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { for probe in &probes { let want = reference.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, @@ -228,6 +232,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { assert_eq!( dpdk.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, @@ -261,6 +266,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { let all_miss: Vec = (0..33u8) .map(|i| LookupInput { src_vpcd: bogus_vpcd(), + dst_vpcd: None, src_ip: format!("10.0.0.{i}").parse().unwrap(), dst_ip: "10.0.0.99".parse().unwrap(), proto: NextHeader::TCP, @@ -314,6 +320,7 @@ fn batched_lookup_matches_single_lookup() { .iter() .map(|p| LookupInput { src_vpcd: p.src_vpcd, + dst_vpcd: None, src_ip: p.src_ip, dst_ip: p.dst_ip, proto: p.proto, @@ -328,6 +335,7 @@ fn batched_lookup_matches_single_lookup() { out[i], tables.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, @@ -354,6 +362,7 @@ fn reference_lookup_matches_config_oracle() { assert_eq!( tables.lookup( probe.src_vpcd, + None, probe.src_ip, probe.dst_ip, probe.proto, diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index d8e98fabf5..e12d12020d 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -78,6 +78,7 @@ pub(crate) enum LookupResult { #[derive(Debug, Clone, Copy)] pub(crate) struct LookupInput { pub(crate) src_vpcd: VpcDiscriminant, + pub(crate) dst_vpcd: Option, pub(crate) src_ip: IpAddr, pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, @@ -95,6 +96,7 @@ pub(super) struct Verdict { /// concrete; every other field is carried verbatim from the [`LookupInput`]). struct Query { src_vni: Vni, + dst_vni: u32, proto: NextHeader, src_ip: I, dst_ip: I, @@ -135,6 +137,9 @@ pub(super) struct RemoteKey { #[exact] #[cli(column_name = "src-vni")] src_vni: Vni, + #[exact] + #[cli(column_name = "dst-vni")] + dst_vni: u32, #[prefix] #[cli(column_name = "destination")] dst_ip: I, @@ -393,6 +398,7 @@ fn emit_remote( v4: &mut Vec, Verdict>>, v6: &mut Vec, Verdict>>, src_vni: Vni, + dst_vni: u32, ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, @@ -407,6 +413,7 @@ fn emit_remote( let rule = RemoteKeyRule:: { proto, src_vni: ExactSpec::new(src_vni), + dst_vni: ExactSpec::new(dst_vni), dst_ip: PrefixSpec::from(prefix), dst_port: port_range, }; @@ -421,6 +428,7 @@ fn emit_remote( let rule = RemoteKeyRule:: { proto, src_vni: ExactSpec::new(src_vni), + dst_vni: ExactSpec::new(dst_vni), dst_ip: PrefixSpec::from(prefix), dst_port: port_range, }; @@ -518,11 +526,17 @@ impl RuleSet { nat_mode: NatRequirement::from_expose(expose), dst_vpcd: remote_vpcd, }; + let dst_vni = if expose.has_masquerade() { + remote_vni.as_u32() + } else { + 0 + }; for prefix in expose.public_ips() { emit_remote( &mut rules.remote_v4, &mut rules.remote_v6, src_vni, + dst_vni, prefix.prefix(), prefix.into(), proto, @@ -535,6 +549,7 @@ impl RuleSet { &mut rules.remote_v4, &mut rules.remote_v6, src_vni, + 0, default_ip(), PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), @@ -640,25 +655,42 @@ impl FlowFilterContext { pub(super) fn lookup( &self, src_vpcd: VpcDiscriminant, + dst_vpcd: Option, src_ip: IpAddr, dst_ip: IpAddr, proto: NextHeader, ports: Option<(u16, u16)>, ) -> LookupResult { let src_vni = key_vni(src_vpcd); + let dst_vni = dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); let (src_port, dst_port) = ports.unzip(); let src_port = src_port.unwrap_or(0); let dst_port = dst_port.unwrap_or(0); match (src_ip, dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { - let Some(verdict) = self.remote_v4.lookup(&RemoteKey { + let verdict = if let Some(v) = self.remote_v4.lookup(&RemoteKey { proto, src_vni, + dst_vni, dst_ip, dst_port, - }) else { - return LookupResult::DestinationMiss; + }) { + v + } else { + if dst_vni != 0 + && let Some(v) = self.remote_v4.lookup(&RemoteKey { + proto, + src_vni, + dst_vni: 0, + dst_ip, + dst_port, + }) + { + v + } else { + return LookupResult::DestinationMiss; + } }; match self.local_v4.lookup(&LocalKey { proto, @@ -674,13 +706,28 @@ impl FlowFilterContext { } } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { - let Some(verdict) = self.remote_v6.lookup(&RemoteKey { + let verdict = if let Some(v) = self.remote_v6.lookup(&RemoteKey { proto, src_vni, + dst_vni, dst_ip, dst_port, - }) else { - return LookupResult::DestinationMiss; + }) { + v + } else { + if dst_vni != 0 + && let Some(v) = self.remote_v6.lookup(&RemoteKey { + proto, + src_vni, + dst_vni: 0, + dst_ip, + dst_port, + }) + { + v + } else { + return LookupResult::DestinationMiss; + } }; match self.local_v6.lookup(&LocalKey { proto, @@ -721,12 +768,14 @@ impl FlowFilterContext { out[i] = LookupResult::DestinationMiss; let proto = input.proto; let src_vni = key_vni(input.src_vpcd); + let dst_vni = input.dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); let (src_port, dst_port) = input.ports.unwrap_or((0, 0)); match (input.src_ip, input.dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { v4_idx.push(i); v4_q.push(Query { src_vni, + dst_vni, proto, src_ip, dst_ip, @@ -738,6 +787,7 @@ impl FlowFilterContext { v6_idx.push(i); v6_q.push(Query { src_vni, + dst_vni, proto, src_ip, dst_ip, @@ -764,7 +814,7 @@ fn lookup_versioned( idx: &[usize], out: &mut [LookupResult], ) where - RemoteKey: MatchKey, + RemoteKey: MatchKey + std::fmt::Debug, LocalKey: MatchKey, { for (q_chunk, i_chunk) in queries.chunks(MAX_BATCH).zip(idx.chunks(MAX_BATCH)) { @@ -774,6 +824,7 @@ fn lookup_versioned( .map(|q| RemoteKey { proto: q.proto, src_vni: q.src_vni, + dst_vni: q.dst_vni, dst_ip: q.dst_ip, dst_port: q.dst_port, }) @@ -781,6 +832,32 @@ fn lookup_versioned( let mut verdicts: Vec> = vec![None; q_chunk.len()]; remote.lookup_batch(&remote_keys, &mut verdicts); + // Reply traffic for masqueraded flows use the destination VNI as part of the key; this is + // to avoid conflicting entries if there are several VPCs exposing overlapping, masqueraded + // prefixes to a given VPC. If we have a destination VNI set here, we may be trying to + // re-validated a reply packet for a masqueraded flow (we're not sure of the direction, + // hence the first attempt with the destination VNI set to 0 above). Try again after setting + // the destination VNI. + let mut reval_positions = Vec::new(); + let mut reval_keys = Vec::new(); + for (pos, (query, verdict)) in q_chunk.iter().zip(verdicts.iter_mut()).enumerate() { + if verdict.is_none() && query.dst_vni != 0 { + reval_positions.push(pos); + reval_keys.push(RemoteKey { + proto: query.proto, + src_vni: query.src_vni, + dst_vni: 0, + dst_ip: query.dst_ip, + dst_port: query.dst_port, + }); + } + } + let mut reval_verdicts = vec![None; reval_keys.len()]; + remote.lookup_batch(&reval_keys, &mut reval_verdicts); + for (pos, verdict) in reval_positions.into_iter().zip(reval_verdicts) { + verdicts[pos] = verdict; + } + // Stage 2: for the hits only, source -> source NAT. let mut local_keys: Vec> = Vec::new(); let mut hit_pos: Vec = Vec::new(); @@ -832,8 +909,8 @@ mod unit_tests { } #[test] - fn remote_key_has_four_fields_local_has_five() { - assert_eq!(RemoteKey::::N, 4); + fn remote_key_has_five_fields_local_has_five_too() { + assert_eq!(RemoteKey::::N, 5); assert_eq!(LocalKey::::N, 5); } diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index c94dcd49bb..c211f637aa 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -28,6 +28,17 @@ fn route( context: &FlowFilterContext, src_vpcd: VpcDiscriminant, headers: &Headers, +) -> Option { + route_revalidate(context, src_vpcd, None, headers) +} + +// Extract the 5-tuple from headers (as the pipeline does) and run the route lookup for a packet +// originating from a given source VPC. +fn route_revalidate( + context: &FlowFilterContext, + src_vpcd: VpcDiscriminant, + dst_vpcd: Option, + headers: &Headers, ) -> Option { let net = headers.net().unwrap(); let src_ip = net.src_addr(); @@ -38,7 +49,7 @@ fn route( .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }); - match context.lookup(src_vpcd, src_ip, dst_ip, proto, ports) { + match context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports) { LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { dst_vpcd, dst_nat, @@ -287,9 +298,10 @@ fn dst_side_nat_modes() { // Masquerade destination: resolves at table level as a marker (the NF only lets it through // for reply traffic on an established masquerade flow; see crate::tests) - let masq = route( + let masq = route_revalidate( &ctx, vpcd(100), + Some(vpcd(200)), &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), ) .expect("masquerade destination resolves as a marker"); @@ -716,8 +728,8 @@ fn reference_and_dpdk_backends_agree() { for &(vni, src_ip, dst_ip, proto, ports) in probes { let src_vpcd = vpcd(vni); assert_eq!( - reference.lookup(src_vpcd, src_ip, dst_ip, proto, ports), - dpdk.lookup(src_vpcd, src_ip, dst_ip, proto, ports), + reference.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports), + dpdk.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports), "backends disagree on {src_ip} -> {dst_ip} ({proto:?}) from vni {vni}", ); } @@ -729,6 +741,7 @@ fn reference_and_dpdk_backends_agree() { .flatten() .map(|&(vni, src_ip, dst_ip, proto, ports)| LookupInput { src_vpcd: vpcd(vni), + dst_vpcd: None, src_ip, dst_ip, proto, @@ -746,6 +759,7 @@ fn reference_and_dpdk_backends_agree() { for (i, input) in inputs.iter().enumerate() { let single = reference.lookup( input.src_vpcd, + input.dst_vpcd, input.src_ip, input.dst_ip, input.proto, @@ -825,6 +839,7 @@ fn display_is_identical_across_backends() { "rank", "proto", "src-vni", + "dst-vni", "destination", "dst-port", "|", @@ -839,6 +854,7 @@ fn display_is_identical_across_backends() { "[0]", "TCP", "100", + "0", "80.0.0.5/32", "2222", "|", @@ -861,10 +877,10 @@ fn display_is_identical_across_backends() { // The heading assertions above would still pass if a section ran past its own table, since they // read its first heading row and stop. The ordering assertion below would not: it compares // positions, so a section that swallowed the table after it could order two rules that are not - // even in the same table and call the result precedence. Pin the boundary directly -- `dst-vni` + // even in the same table and call the result precedence. Pin the boundary directly -- `source` // is a local-table column, and the local table is the one that follows. assert!( - !remote_v4.contains("dst-vni"), + !remote_v4.contains("source"), "the remote v4 section ran past its own table:\n{remote_v4}" ); diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index 7aa534d111..d1bfbc7fbe 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -118,6 +118,7 @@ impl FlowFilter { genid: i64, ) -> Classification { let nfi = &self.name; + let mut revalidation_dst_vpcd = None; let attached_flow = FlowSummary::from_meta(packet.meta()); if let Some(flow_summary) = attached_flow.as_ref() { // Bypass flow-filter if packet has up-to-date active flow-info @@ -125,6 +126,7 @@ impl FlowFilter { Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow_summary); return Classification::Bypassed; } + revalidation_dst_vpcd = self.flow_revalidation_data(flow_summary, genid); } let Some(net) = packet.try_ip() else { @@ -140,6 +142,7 @@ impl FlowFilter { let input = LookupInput { src_vpcd, + dst_vpcd: revalidation_dst_vpcd, src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -156,14 +159,6 @@ impl FlowFilter { } /// Phase C: apply a resolved route (or drop on a miss) to a single packet. - /// - /// The tables cannot answer for reply traffic of established stateful-NAT sessions: masquerade - /// destinations only appear as marker rules (they cannot accept new connections) and - /// port-forwarding sources are absent altogether (they cannot initiate). For those two cases - /// -- and only those -- an active flow carrying the matching NAT state lets the packet - /// through, exactly as the flow-bypass path would. The flow's validity under the new - /// configuration remains the stateful NFs' responsibility; a genuine miss (no peering covers - /// the packet) still drops and invalidates. fn apply_route( &self, packet: &mut Packet, @@ -196,25 +191,6 @@ impl FlowFilter { return; } }; - - // A masquerade destination cannot accept new connections; its rule is in the table only - // so that reply traffic on an established masquerade flow is distinguishable from a - // destination no peering covers. - if dst_nat_mode == Some(NatRequirement::Masquerade) { - if let Some(flow) = active_stateful_flow(flow_summary, dst_vpcd, |f| f.needs_masquerade) - { - debug!("{nfi}: Masquerade destination allowed by established flow"); - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow); - return; - } - debug!( - "{nfi}: Masquerade destination with no established flow, dropping packet (cannot initiate a connection towards a masquerade expose)" - ); - packet.invalidate_flows(); - packet.done(DoneReason::Filtered); - return; - } - debug!( "{nfi}: Packet matches peering configuration, found VPC {dst_vpcd} and NAT modes {src_nat_mode:?} (src), {dst_nat_mode:?} (dst)" ); @@ -324,6 +300,21 @@ impl FlowFilter { false } + fn flow_revalidation_data( + &self, + flow_summary: &FlowSummary, + genid: i64, + ) -> Option { + // The only case when we need re-validation is when we have an active flow with an outdated + // genid. If this is not the case, return None. + if flow_summary.flow_info.status() != FlowStatus::Active + || flow_summary.flow_info.genid() >= genid + { + return None; + } + flow_summary.dst_vpcd + } + fn dst_vpcd_from_valid_flow( &self, flow_summary: &FlowSummary, From 086f8edf8dc0bc5dca84db3ca04d6a67aa667e6a Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 03:16:11 +0100 Subject: [PATCH 2/8] feat(flow-filter): Key port-forwarding sources on the NAT mode A port-forwarding source cannot initiate a connection, so it was left out of the local table altogether. The reply direction of a forwarded session then gets no answer from the tables, when we try to re-validate it after a configuration update: it comes back as a source miss. Instead, add back the related entries to the local table; but also make the NAT mode part of the key to use it in that case (and only in the case of reply traffic for port-forwarding flows): we need it to dissociate these entries from potential colliding entries associated with overlapping masqueraded prefixes. Similarly to what we do for the remote lookup table, we have a two-step lookup: first look with the NAT mode, to catch an entry associated with port-forwarding reply traffic, then retry without it for "forward" traffic. Signed-off-by: Quentin Monnet --- flow-filter/src/context/fuzz.rs | 17 +++- flow-filter/src/context/tables.rs | 106 +++++++++++++++++++---- flow-filter/src/context/tests.rs | 135 ++++++++++++++++++++++-------- flow-filter/src/lib.rs | 63 +++++++------- flow-filter/src/tests.rs | 10 +-- 5 files changed, 239 insertions(+), 92 deletions(-) diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index 31dcd771fb..d82d9d2f31 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -183,6 +183,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { probe.dst_ip, probe.proto, probe.ports, + None, ); assert_eq!( dpdk.lookup( @@ -191,7 +192,8 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { probe.src_ip, probe.dst_ip, probe.proto, - probe.ports + probe.ports, + None, ), want, "backends disagree on derived routing probe {probe:?}\nspec: {overlay_spec:?}", @@ -216,6 +218,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { dst_ip: p.dst_ip, proto: p.proto, ports: p.ports, + nat_mode: None, }) .collect(); @@ -228,6 +231,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { probe.dst_ip, probe.proto, probe.ports, + None, ); assert_eq!( dpdk.lookup( @@ -236,7 +240,8 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { probe.src_ip, probe.dst_ip, probe.proto, - probe.ports + probe.ports, + None, ), want, "backends disagree on single lookup of {probe:?}\nspec: {overlay_spec:?}", @@ -271,6 +276,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { dst_ip: "10.0.0.99".parse().unwrap(), proto: NextHeader::TCP, ports: Some((1, 2)), + nat_mode: None, }) .collect(); let mut out = vec![LookupResult::DestinationMiss; all_miss.len()]; @@ -325,6 +331,7 @@ fn batched_lookup_matches_single_lookup() { dst_ip: p.dst_ip, proto: p.proto, ports: p.ports, + nat_mode: None, }) .collect(); @@ -339,7 +346,8 @@ fn batched_lookup_matches_single_lookup() { probe.src_ip, probe.dst_ip, probe.proto, - probe.ports + probe.ports, + None, ), "batch slot {i} != single lookup for {probe:?}\nspec: {overlay_spec:?}", ); @@ -366,7 +374,8 @@ fn reference_lookup_matches_config_oracle() { probe.src_ip, probe.dst_ip, probe.proto, - probe.ports + probe.ports, + None, ), oracle_lookup(&built.overlay, &probe), "reference tables disagree with the config oracle on {probe:?}\nspec: {overlay_spec:?}", diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index e12d12020d..67ee4c193b 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -83,6 +83,7 @@ pub(crate) struct LookupInput { pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, + pub(crate) nat_mode: NatMode, } /// Result of a stage-1 (remote/destination) match. @@ -102,6 +103,7 @@ struct Query { dst_ip: I, src_port: u16, dst_port: u16, + nat_mode: u8, } /// Lower a config L4 protocol to a bitmask predicate: a specific protocol matches exactly (every @@ -165,6 +167,9 @@ pub(super) struct LocalKey { #[range] #[cli(column_name = "src-port")] src_port: u16, + #[exact] + #[cli(column_name = "nat-mode")] + nat_mode: u8, } // ------------------------------------------------------------------------------------------------- @@ -452,6 +457,7 @@ fn emit_local( ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, + nat_mode: NatMode, action: NatMode, ) { // Port-forwarding sources are never emitted into the local tables, so the tie-break bit is @@ -465,6 +471,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, + nat_mode: ExactSpec::new(NatRequirement::convert_option(nat_mode)), }; v4.push(NeutralRule { priority, @@ -480,6 +487,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, + nat_mode: ExactSpec::new(NatRequirement::convert_option(nat_mode)), }; v6.push(NeutralRule { priority, @@ -562,14 +570,14 @@ impl RuleSet { // Stage 2: source's private prefixes -> source NAT mode. Port-forwarding sources // cannot initiate connections, so they are excluded here. - for expose in peering - .local() - .valexp() - .iter() - .filter(|expose| expose.can_init_connection()) - { + for expose in peering.local().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = NatRequirement::from_expose(expose); + let nat_mode = if expose.has_port_forwarding() { + action + } else { + None + }; for prefix in expose.ips() { emit_local( &mut rules.local_v4, @@ -579,6 +587,7 @@ impl RuleSet { prefix.prefix(), prefix.into(), proto, + nat_mode, action, ); } @@ -593,6 +602,7 @@ impl RuleSet { PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), None, + None, ); } } @@ -652,6 +662,7 @@ impl FlowFilterContext { // Single-key lookup: the readable per-packet oracle used by tests; production runs // lookup_batch. The differential test cross-checks the two against each other. #[cfg(test)] + #[allow(clippy::too_many_arguments)] pub(super) fn lookup( &self, src_vpcd: VpcDiscriminant, @@ -660,12 +671,14 @@ impl FlowFilterContext { dst_ip: IpAddr, proto: NextHeader, ports: Option<(u16, u16)>, + nat_mode: NatMode, ) -> LookupResult { let src_vni = key_vni(src_vpcd); let dst_vni = dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); let (src_port, dst_port) = ports.unzip(); let src_port = src_port.unwrap_or(0); let dst_port = dst_port.unwrap_or(0); + let nat_mode = NatRequirement::convert_option(nat_mode); match (src_ip, dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { @@ -692,17 +705,34 @@ impl FlowFilterContext { return LookupResult::DestinationMiss; } }; + let dst_vni = key_vni(verdict.dst_vpcd); match self.local_v4.lookup(&LocalKey { proto, src_vni, - dst_vni: key_vni(verdict.dst_vpcd), + dst_vni, src_ip, src_port, + nat_mode, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) + Some(src_nat_mode) => { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } + None => { + if nat_mode != 0 + && let Some(src_nat_mode) = self.local_v4.lookup(&LocalKey { + proto, + src_vni, + dst_vni, + src_ip, + src_port, + nat_mode: 0, + }) + { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } else { + LookupResult::SourceMiss(verdict.dst_vpcd) + } } - None => LookupResult::SourceMiss(verdict.dst_vpcd), } } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { @@ -729,17 +759,34 @@ impl FlowFilterContext { return LookupResult::DestinationMiss; } }; + let dst_vni = key_vni(verdict.dst_vpcd); match self.local_v6.lookup(&LocalKey { proto, src_vni, - dst_vni: key_vni(verdict.dst_vpcd), + dst_vni, src_ip, src_port, + nat_mode, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) + Some(src_nat_mode) => { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } + None => { + if nat_mode != 0 + && let Some(src_nat_mode) = self.local_v6.lookup(&LocalKey { + proto, + src_vni, + dst_vni, + src_ip, + src_port, + nat_mode: 0, + }) + { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } else { + LookupResult::SourceMiss(verdict.dst_vpcd) + } } - None => LookupResult::SourceMiss(verdict.dst_vpcd), } } _ => { @@ -770,6 +817,7 @@ impl FlowFilterContext { let src_vni = key_vni(input.src_vpcd); let dst_vni = input.dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); let (src_port, dst_port) = input.ports.unwrap_or((0, 0)); + let nat_mode = NatRequirement::convert_option(input.nat_mode); match (input.src_ip, input.dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { v4_idx.push(i); @@ -781,6 +829,7 @@ impl FlowFilterContext { dst_ip, src_port, dst_port, + nat_mode, }); } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { @@ -793,6 +842,7 @@ impl FlowFilterContext { dst_ip, src_port, dst_port, + nat_mode, }); } _ => { /* version mismatch: leave "out[i] = DestinationMiss" */ } @@ -859,6 +909,8 @@ fn lookup_versioned( } // Stage 2: for the hits only, source -> source NAT. + // Port-forwarding rules use the NAT mode as part of the key, to dissociate keys from any + // keys associated to overlapping forward masquerade prefixes. let mut local_keys: Vec> = Vec::new(); let mut hit_pos: Vec = Vec::new(); for (pos, verdict) in verdicts.iter().enumerate() { @@ -870,6 +922,7 @@ fn lookup_versioned( dst_vni: key_vni(verdict.dst_vpcd), src_ip: q.src_ip, src_port: q.src_port, + nat_mode: q.nat_mode, }); hit_pos.push(pos); } @@ -877,6 +930,27 @@ fn lookup_versioned( let mut nat_modes: Vec> = vec![None; local_keys.len()]; local.lookup_batch(&local_keys, &mut nat_modes); + // Second pass: if nat_mode was set and we didn't find an entry for reply traffic associated + // with a port-forwarding flow, set nat_mode to 0 to see if we have an entry for forward + // traffic for port-forwarding (forward traffic entries do not have flow-info nat mode + // attached, or we couldn't use it to initiate new flows). + let mut reval_positions = Vec::new(); + let mut reval_keys = Vec::new(); + for (pos, (nat_mode, &q_pos)) in nat_modes.iter().zip(hit_pos.iter()).enumerate() { + if nat_mode.is_none() && q_chunk[q_pos].nat_mode != 0 { + reval_positions.push(pos); + reval_keys.push(LocalKey { + nat_mode: 0, + ..local_keys[pos].clone() + }); + } + } + let mut reval_nat_modes = vec![None; reval_keys.len()]; + local.lookup_batch(&reval_keys, &mut reval_nat_modes); + for (pos, nat_mode) in reval_positions.into_iter().zip(reval_nat_modes) { + nat_modes[pos] = nat_mode; + } + // Scatter results back to the caller's output positions. A stage-1 miss stays // DestinationMiss; a stage-1 hit whose source matched nothing becomes SourceMiss. for (hit, &pos) in hit_pos.iter().enumerate() { @@ -909,9 +983,9 @@ mod unit_tests { } #[test] - fn remote_key_has_five_fields_local_has_five_too() { + fn remote_key_has_five_fields_local_has_six() { assert_eq!(RemoteKey::::N, 5); - assert_eq!(LocalKey::::N, 5); + assert_eq!(LocalKey::::N, 6); } #[test] diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index c211f637aa..cab533a58b 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -29,17 +29,25 @@ fn route( src_vpcd: VpcDiscriminant, headers: &Headers, ) -> Option { - route_revalidate(context, src_vpcd, None, headers) + match route_lookup(context, src_vpcd, None, None, headers) { + LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { + dst_vpcd, + dst_nat, + src_nat, + }), + LookupResult::SourceMiss(_) | LookupResult::DestinationMiss => None, + } } // Extract the 5-tuple from headers (as the pipeline does) and run the route lookup for a packet // originating from a given source VPC. -fn route_revalidate( +fn route_lookup( context: &FlowFilterContext, src_vpcd: VpcDiscriminant, dst_vpcd: Option, + nat_mode: NatMode, headers: &Headers, -) -> Option { +) -> LookupResult { let net = headers.net().unwrap(); let src_ip = net.src_addr(); let dst_ip = net.dst_addr(); @@ -49,14 +57,7 @@ fn route_revalidate( .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }); - match context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports) { - LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { - dst_vpcd, - dst_nat, - src_nat, - }), - LookupResult::SourceMiss(_) | LookupResult::DestinationMiss => None, - } + context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports, nat_mode) } // ------------------------------------------------------------------------------------------------- @@ -197,8 +198,7 @@ fn overlapping_source_prefix_disambiguated_by_destination() { // We pin down which NAT requirement is returned for each end of a lookup. The source (local) end // carries private IPs; the destination (remote) end carries public IPs. Masquerade is only valid on // the source side (a masquerade destination cannot receive connections) and port forwarding only on -// the destination side (a port-forwarding source cannot initiate connections); these constraints -// are tested in `dst_side_nat_modes`. +// the destination side (a port-forwarding source cannot initiate connections). fn nat_modes_overlay() -> FlowFilterContext { context( @@ -296,17 +296,44 @@ fn dst_side_overlay() -> FlowFilterContext { fn dst_side_nat_modes() { let ctx = dst_side_overlay(); - // Masquerade destination: resolves at table level as a marker (the NF only lets it through - // for reply traffic on an established masquerade flow; see crate::tests) - let masq = route_revalidate( + // Masquerade source + let masq = route( + &ctx, + vpcd(200), + &build_tcp_packet(v4("192.168.70.1"), v4("10.0.0.5"), 1234, 5678), + ) + .expect("masquerade destination resolves as a marker"); + assert_eq!(masq.dst_vpcd, vpcd(100)); + assert_eq!(masq.dst_nat, None); + assert_eq!(masq.src_nat, Some(NatRequirement::Masquerade)); + + // Masquerade destination: without the destination VPC discriminant hint, we fail to find the + // relevant destination entry. This is expected, because we can never initiate a flow in this + // direction, so we need the dst_vpcd from flow info to find the relevant entry (only necessary + // when re-validating after a configuration change). + let lookup_result = route_lookup( + &ctx, + vpcd(100), + None, + None, + &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), + ); + assert_eq!(lookup_result, LookupResult::DestinationMiss); + + // Masquerade destination: With the destination VPC discriminant (virtually-)retrieved from flow + // information, we can determinate the right information for the packet. + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( &ctx, vpcd(100), Some(vpcd(200)), + None, &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), - ) - .expect("masquerade destination resolves as a marker"); - assert_eq!(masq.dst_vpcd, vpcd(200)); - assert_eq!(masq.dst_nat, Some(NatRequirement::Masquerade)); + ) else { + panic!("masquerade destination resolves as a marker"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(dst_nat, Some(NatRequirement::Masquerade)); + assert_eq!(src_nat, None); // Port-forwarding destination (matching proto + port): returned let pf = route( @@ -319,6 +346,30 @@ fn dst_side_nat_modes() { assert_eq!(pf.dst_nat, Some(NatRequirement::PortForwarding)); assert_eq!(pf.src_nat, None); + // Port-forwarding source without NAT mode hint: lookup fails + let lookup_result = route_lookup( + &ctx, + vpcd(200), + None, + None, + &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), + ); + assert_eq!(lookup_result, LookupResult::SourceMiss(vpcd(100))); + + // Port-forwarding source without NAT mode hint: lookup fails + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( + &ctx, + vpcd(200), + None, + Some(NatRequirement::PortForwarding), + &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), + ) else { + panic!("masquerade destination resolves as a marker"); + }; + assert_eq!(dst_vpcd, vpcd(100)); + assert_eq!(dst_nat, None); + assert_eq!(src_nat, Some(NatRequirement::PortForwarding)); + // Port-forwarding destination, wrong port: no match. assert_eq!( route( @@ -440,12 +491,12 @@ fn port_forwarding_any_protocol_matches_tcp_and_udp() { } // ------------------------------------------------------------------------------------------------- -// Port forwarding is excluded from the source side (it cannot initiate connections). With both a -// masquerade and a port-forwarding expose on the source manifest, a source in the port-forwarding -// range is matched by masquerade instead. +// Port-forwarding and masquerade prefixes may overlap within a manifest. In this case, for +// re-validating return traffic for port-forwarding, we rely on the NAT mode from the flow +// information to make the distinction with similar-looking entries for forward masqueraded traffic. #[test] -fn source_port_forwarding_is_excluded_and_falls_back_to_masquerade() { +fn port_forwarding_and_masquerade_overlap_resoves_as_expected() { let ctx = context( &[("vpc1", 100), ("vpc2", 200)], vec![peering( @@ -467,15 +518,31 @@ fn source_port_forwarding_is_excluded_and_falls_back_to_masquerade() { )], ); // Source 1.0.0.27:2000 is inside the port-forwarding private range, yet resolves to masquerade. - let r = route( + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( &ctx, vpcd(100), + None, + Some(NatRequirement::PortForwarding), &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), - ) - .expect("source resolves via the masquerade expose"); - assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.src_nat, Some(NatRequirement::Masquerade)); - assert_eq!(r.dst_nat, None); + ) else { + panic!("source resolves via the port-forwarding expose"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(src_nat, Some(NatRequirement::PortForwarding)); + assert_eq!(dst_nat, None); + + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( + &ctx, + vpcd(100), + None, + None, + &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), + ) else { + panic!("source resolves via the masquerade expose"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(src_nat, Some(NatRequirement::Masquerade)); + assert_eq!(dst_nat, None); } // ------------------------------------------------------------------------------------------------- @@ -728,8 +795,8 @@ fn reference_and_dpdk_backends_agree() { for &(vni, src_ip, dst_ip, proto, ports) in probes { let src_vpcd = vpcd(vni); assert_eq!( - reference.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports), - dpdk.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports), + reference.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports, None), + dpdk.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports, None), "backends disagree on {src_ip} -> {dst_ip} ({proto:?}) from vni {vni}", ); } @@ -746,6 +813,7 @@ fn reference_and_dpdk_backends_agree() { dst_ip, proto, ports, + nat_mode: None, }) .collect(); assert!(inputs.len() > 32, "want a multi-chunk batch"); @@ -764,6 +832,7 @@ fn reference_and_dpdk_backends_agree() { input.dst_ip, input.proto, input.ports, + input.nat_mode, ); assert_eq!(ref_out[i], single, "batched != single at index {i}"); } @@ -869,7 +938,7 @@ fn display_is_identical_across_backends() { assert_eq!( cells(&local_v4, "rank"), [ - "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "|", "NAT" + "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "nat-mode", "|", "NAT" ], "unexpected local heading row:\n{local_v4}" ); diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index d1bfbc7fbe..aa011d543b 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -118,7 +118,7 @@ impl FlowFilter { genid: i64, ) -> Classification { let nfi = &self.name; - let mut revalidation_dst_vpcd = None; + let (mut revalidation_dst_vpcd, mut revalidation_nat_mode) = (None, None); let attached_flow = FlowSummary::from_meta(packet.meta()); if let Some(flow_summary) = attached_flow.as_ref() { // Bypass flow-filter if packet has up-to-date active flow-info @@ -126,7 +126,8 @@ impl FlowFilter { Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow_summary); return Classification::Bypassed; } - revalidation_dst_vpcd = self.flow_revalidation_data(flow_summary, genid); + (revalidation_dst_vpcd, revalidation_nat_mode) = + self.flow_revalidation_data(flow_summary, genid); } let Some(net) = packet.try_ip() else { @@ -151,6 +152,7 @@ impl FlowFilter { .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }), + nat_mode: revalidation_nat_mode, }; Classification::Lookup { input, @@ -170,15 +172,6 @@ impl FlowFilter { let (dst_vpcd, dst_nat_mode, src_nat_mode) = match result { LookupResult::Route(route) => route, LookupResult::SourceMiss(dst_vpcd) => { - // Port-forwarding sources are deliberately absent from the local tables; reply - // traffic from one rides its established flow. - if let Some(flow) = - active_stateful_flow(flow_summary, dst_vpcd, |f| f.needs_port_forwarding) - { - debug!("{nfi}: Source allowed by established port-forwarding flow"); - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow); - return; - } debug!("{nfi}: Source not allowed towards {dst_vpcd}, dropping packet"); packet.invalidate_flows(); packet.done(DoneReason::Filtered); @@ -304,15 +297,25 @@ impl FlowFilter { &self, flow_summary: &FlowSummary, genid: i64, - ) -> Option { + ) -> (Option, Option) { // The only case when we need re-validation is when we have an active flow with an outdated // genid. If this is not the case, return None. if flow_summary.flow_info.status() != FlowStatus::Active || flow_summary.flow_info.genid() >= genid { - return None; + return (None, None); + } + if flow_summary.needs_masquerade { + // If we need revalidation and the flow is masqueraded, we may need the destination VPC + // id from the flow to look up for reverse traffic's entry in the "remote" table. + (flow_summary.dst_vpcd, None) + } else if flow_summary.needs_port_forwarding { + // In we need revalidation and the flow is with port-forwarding, we may need the NAT + // mode from the flow to look up for reverse traffic's entry in the "local" table. + (None, Some(NatRequirement::PortForwarding)) + } else { + (None, None) } - flow_summary.dst_vpcd } fn dst_vpcd_from_valid_flow( @@ -391,23 +394,6 @@ impl FlowSummary { } } -/// The flow, if it is active, agrees with the lookup on the destination VPC, and carries the -/// stateful-NAT state selected by `has_state`. Such a flow vouches for reply traffic that the -/// tables cannot answer for (see [`FlowFilter::apply_route`]). No genid check: an up-to-date flow -/// would have bypassed the lookup already, and an outdated one is exactly the case where the flow -/// must speak for the packet; the stateful NFs remain the authority on the state itself. -fn active_stateful_flow( - flow_summary: Option<&FlowSummary>, - dst_vpcd: VpcDiscriminant, - has_state: impl Fn(&FlowSummary) -> bool, -) -> Option<&FlowSummary> { - flow_summary.filter(|flow| { - flow.flow_info.status() == FlowStatus::Active - && flow.dst_vpcd == Some(dst_vpcd) - && has_state(flow) - }) -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum NatRequirement { Static, @@ -423,6 +409,21 @@ impl NatRequirement { VpcExposeNatConfig::PortForwarding(_) => Some(Self::PortForwarding), } } + + fn as_u8(&self) -> u8 { + match self { + NatRequirement::Static => 1, + NatRequirement::Masquerade => 2, + NatRequirement::PortForwarding => 3, + } + } + + fn convert_option(opt: Option) -> u8 { + match opt { + Some(r) => r.as_u8(), + None => 0, + } + } } pub(crate) type NatMode = Option; diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index cb9222387b..152cb18d4e 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -566,11 +566,7 @@ fn outdated_flow_with_consistent_state_is_kept() { } // ------------------------------------------------------------------------------------------------- -// Stateful reply traffic across config changes. The tables cannot answer for the reverse direction -// of stateful-NAT sessions (masquerade destinations are only markers, port-forwarding sources are -// absent altogether), so after a genid bump those packets must ride their established flow instead -// of being dropped -- while packets with no such flow, and flows whose peering is gone, still fail -// closed. +// Stateful reply traffic across config changes. #[test] fn masquerade_reply_on_established_flow_survives_config_change() { @@ -626,7 +622,6 @@ fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { Some(vpcd(200)), build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), ); - // The flow's recorded destination does not match what the tables resolve: stale, drop. let flow = attach_flow(&mut p, Some(vpcd(300)), true, true, false); let out = run(&mut flow_filter, p); assert_eq!(out.get_done(), Some(DoneReason::Filtered)); @@ -637,8 +632,7 @@ fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { fn port_forwarding_reply_on_established_flow_survives_config_change() { let (mut flow_filter, _) = make_flow_filter(dst_port_forwarding_context()); set_genid(&mut flow_filter, 5); - // Reply direction of a forwarded session: the forwarded host answers from its private - // address, which is (deliberately) not in the local tables. + // Reply direction of a forwarded session: the forwarded host answers from its private address. let mut p = packet( Some(vpcd(200)), build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), From e26f6dc7b335fb5190297437bb16cc1f542f2d5f Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 03:56:42 +0100 Subject: [PATCH 3/8] test(flow-filter): Update fuzzing tests for flow re-validation logic The tables now hold rules that only a lookup carrying flow revalidation information can reach. The property tests still asked every question without it, so the config oracle described semantics the tables no longer have, and the probes derived to route by construction stopped routing. Those derived probes are also the only ones that reach the gated rules at all: the address, port and NAT mode a revalidated route needs never come up together by random generation. They now go through the batch path and the oracle too, not just the single lookup, and the run fails if it never resolved a route through revalidation. Also fix some comments still describing masquerade destinations and port-forwarding sources as marker-only or absent from the tables, and document the ordering. (Initially written by Claude, then rebased by Claude on top of the recent flow-filter fuzz tests changes from Daniel.) Co-authored-by: Claude Opus 5 (1M context) Signed-off-by: Quentin Monnet --- flow-filter/src/context/fuzz.rs | 301 ++++++++++++++++-------------- flow-filter/src/context/tables.rs | 36 ++-- flow-filter/src/fuzz_gen.rs | 176 +++++++++++++---- flow-filter/src/tests.rs | 8 + 4 files changed, 336 insertions(+), 185 deletions(-) diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index d82d9d2f31..ca56054397 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -12,11 +12,12 @@ #![cfg(test)] use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult}; -use crate::NatRequirement; use crate::fuzz_gen::{OverlaySpec, Probe, ProbeSpec, bogus_vpcd}; +use crate::{NatMode, NatRequirement}; use concurrency::sync::LazyLock; use concurrency::sync::atomic::{AtomicU64, Ordering}; use config::external::overlay::ValidatedOverlay; +use config::external::overlay::vpc::{ValidatedPeering, ValidatedVpc}; use lpm::prefix::{IpPrefix, L4Protocol, Prefix, PrefixWithOptionalPorts}; use net::ip::NextHeader; use net::packet::VpcDiscriminant; @@ -66,28 +67,31 @@ fn consider(best: &mut Option<(Precedence, T)>, precedence: Precedence, value } } -/// Answer a route lookup directly from the validated overlay. -/// 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() - .find(|vpc| VpcDiscriminant::from_vni(vpc.vni()) == probe.src_vpcd) - else { - return LookupResult::DestinationMiss; - }; - if probe.src_ip.is_ipv4() != probe.dst_ip.is_ipv4() { - return LookupResult::DestinationMiss; - } - let (sport, dport) = probe.ports.unwrap_or((0, 0)); - - // Stage 1: the destination against every peer's public prefixes. Masquerade exposes are - // included (marker rules); a default expose acts as a /0 of the peering's IP version. +/// Stage 1: the destination against every peer's public prefixes, scoped to the source VPC. +/// +/// `revalidated` is the destination VPC an outdated flow vouches for. With it, only the masquerade +/// exposes of the peering to that VPC answer -- they cannot receive connections, so nothing but a +/// flow that already used them can reach them. Without it, only exposes that can receive do (a +/// default expose acts as a /0 of the peering's IP version). +fn oracle_stage1( + src_vpc: &ValidatedVpc, + probe: &Probe, + dport: u16, + revalidated: Option, +) -> Option<(VpcDiscriminant, Option)> { let mut verdict: Option<(Precedence, (VpcDiscriminant, Option))> = None; for peering in src_vpc.peerings() { let dst_vpcd = VpcDiscriminant::from_vni(peering.remote_vni()); + if revalidated.is_some_and(|vpcd| vpcd != dst_vpcd) { + continue; + } for expose in peering.remote().valexp() { - if !proto_allows(expose.nat_proto(), probe.proto) { + let matchable = if revalidated.is_some() { + expose.has_masquerade() + } else { + expose.can_receive_connection() + }; + if !matchable || !proto_allows(expose.nat_proto(), probe.proto) { continue; } for prefix in expose.public_ips() { @@ -100,29 +104,37 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup } } } - if peering.remote().has_default_expose() && probe.dst_ip.is_ipv4() == peering.is_v4() { + if revalidated.is_none() + && peering.remote().has_default_expose() + && probe.dst_ip.is_ipv4() == peering.is_v4() + { consider(&mut verdict, (0, false), (dst_vpcd, None)); } } - let Some((_, (dst_vpcd, dst_nat))) = verdict else { - return LookupResult::DestinationMiss; - }; + verdict.map(|(_, hit)| hit) +} - // Stage 2: the source against that peering's private prefixes. Port-forwarding sources are - // excluded (they cannot initiate); a default expose acts as a /0 of the peering's version. - let peering = src_vpc - .peerings() - .iter() - .find(|p| VpcDiscriminant::from_vni(p.remote_vni()) == dst_vpcd) - .unwrap_or_else(|| unreachable!("stage 1 hit implies a peering to the verdict VPC")); +/// Stage 2: the source against the resolved peering's private prefixes. +/// +/// `revalidated` is the source NAT mode an outdated flow vouches for. With it, only the +/// port-forwarding exposes of that very mode answer -- they cannot initiate connections, so +/// nothing but a flow that already used them can reach them. Without it, only exposes that can +/// initiate do (a default expose acts as a /0 of the peering's IP version). +fn oracle_stage2( + peering: &ValidatedPeering, + probe: &Probe, + sport: u16, + revalidated: NatMode, +) -> Option> { let mut src_nat: Option<(Precedence, Option)> = None; - for expose in peering - .local() - .valexp() - .iter() - .filter(|expose| expose.can_init_connection()) - { - if !proto_allows(expose.nat_proto(), probe.proto) { + for expose in peering.local().valexp() { + let matchable = match revalidated { + Some(mode) => { + expose.has_port_forwarding() && NatRequirement::from_expose(expose) == Some(mode) + } + None => expose.can_init_connection(), + }; + if !matchable || !proto_allows(expose.nat_proto(), probe.proto) { continue; } for prefix in expose.ips() { @@ -135,11 +147,54 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup } } } - if peering.local().has_default_expose() && probe.src_ip.is_ipv4() == peering.is_v4() { + if revalidated.is_none() + && peering.local().has_default_expose() + && probe.src_ip.is_ipv4() == peering.is_v4() + { consider(&mut src_nat, (0, false), None); } + src_nat.map(|(_, nat)| nat) +} + +/// Answer a route lookup directly from the validated overlay. +/// Shared by the context and NF metadata property tests. +/// +/// Both stages ask what the flow vouches for first, and only then the plain question. A packet on +/// an established flow keeps the peering and the NAT mode that flow was built on, even where an +/// expose covers the same address ungated; the plain question is the fallback, for the forward +/// traffic that carries no flow information of its own. +pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { + let Some(src_vpc) = overlay + .vpc_table() + .values() + .find(|vpc| VpcDiscriminant::from_vni(vpc.vni()) == probe.src_vpcd) + else { + return LookupResult::DestinationMiss; + }; + if probe.src_ip.is_ipv4() != probe.dst_ip.is_ipv4() { + return LookupResult::DestinationMiss; + } + let (sport, dport) = probe.ports.unwrap_or((0, 0)); + + let verdict = probe + .dst_vpcd + .and_then(|vpcd| oracle_stage1(src_vpc, probe, dport, Some(vpcd))) + .or_else(|| oracle_stage1(src_vpc, probe, dport, None)); + let Some((dst_vpcd, dst_nat)) = verdict else { + return LookupResult::DestinationMiss; + }; + + let peering = src_vpc + .peerings() + .iter() + .find(|p| VpcDiscriminant::from_vni(p.remote_vni()) == dst_vpcd) + .unwrap_or_else(|| unreachable!("stage 1 hit implies a peering to the verdict VPC")); + let src_nat = probe + .nat_mode + .and_then(|mode| oracle_stage2(peering, probe, sport, Some(mode))) + .or_else(|| oracle_stage2(peering, probe, sport, None)); match src_nat { - Some((_, src_nat)) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), + Some(src_nat) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), None => LookupResult::SourceMiss(dst_vpcd), } } @@ -147,6 +202,32 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup // ------------------------------------------------------------------------------------------------- // Properties. +/// A probe's fields are exactly one lookup's arguments, revalidation information included. +fn lookup(tables: &FlowFilterContext, probe: &Probe) -> LookupResult { + tables.lookup( + probe.src_vpcd, + probe.dst_vpcd, + probe.src_ip, + probe.dst_ip, + probe.proto, + probe.ports, + probe.nat_mode, + ) +} + +/// The same question, in the batch path's form. +fn lookup_input(probe: &Probe) -> LookupInput { + LookupInput { + src_vpcd: probe.src_vpcd, + dst_vpcd: probe.dst_vpcd, + src_ip: probe.src_ip, + dst_ip: probe.dst_ip, + proto: probe.proto, + ports: probe.ports, + nat_mode: probe.nat_mode, + } +} + /// The rte_acl backend agrees with the reference backend on every probe of every generated /// overlay -- single lookups and the chunked batch path alike. This is the fuzz form of /// `tests::reference_and_dpdk_backends_agree`: it validates the wide-key encoding and the @@ -160,6 +241,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { // Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not // const (each instance registers with the loom executor). static ROUTES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static REVALIDATED_ROUTES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static SOURCE_MISSES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DESTINATION_MISSES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -171,78 +253,34 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { .expect("reference build"); let dpdk = FlowFilterContext::build(&built.overlay, Backend::Dpdk).expect("dpdk build"); - // Guaranteed-routing probes derived from the overlay's own structure. A full route from - // random generation would be a rare random outcome (~1% of generated routes), and we - // may not get enough of them during a time-boxed bolero run on a loaded CI runner to - // get meaningful coverage. + // Probes derived from the overlay's own structure route by construction. A full route + // from random generation would be a rare random outcome (~1% of generated routes), and + // we may not get enough of them during a time-boxed bolero run on a loaded CI runner to + // get meaningful coverage -- the revalidated ones would get none at all. for probe in &built.routing_probes { - let want = reference.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ); - assert_eq!( - dpdk.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ), - want, - "backends disagree on derived routing probe {probe:?}\nspec: {overlay_spec:?}", - ); assert!( - matches!(want, LookupResult::Route(_)), + matches!(lookup(&reference, probe), LookupResult::Route(_)), "derived routing probe did not route: {probe:?}\nspec: {overlay_spec:?}", ); - ROUTES.fetch_add(1, Ordering::Relaxed); + if probe.dst_vpcd.is_some() || probe.nat_mode.is_some() { + REVALIDATED_ROUTES.fetch_add(1, Ordering::Relaxed); + } } - let probes: Vec = probe_specs - .iter() - .map(|p| p.resolve(built.blocks)) - .collect(); - let inputs: Vec = probes + // Derived and random probes alike go through both backends and both lookup paths. + let probes: Vec = built + .routing_probes .iter() - .map(|p| LookupInput { - src_vpcd: p.src_vpcd, - dst_vpcd: None, - src_ip: p.src_ip, - dst_ip: p.dst_ip, - proto: p.proto, - ports: p.ports, - nat_mode: None, - }) + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))) .collect(); + let inputs: Vec = probes.iter().map(lookup_input).collect(); let mut expected = Vec::with_capacity(probes.len()); for probe in &probes { - let want = reference.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ); + let want = lookup(&reference, probe); assert_eq!( - dpdk.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ), + lookup(&dpdk, probe), want, "backends disagree on single lookup of {probe:?}\nspec: {overlay_spec:?}", ); @@ -256,7 +294,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { expected.push(want); } - // Batch path: 40 inputs > MAX_BATCH exercises the chunked scatter; every slot must + // Batch path: more than MAX_BATCH inputs exercise the chunked scatter; every slot must // equal the corresponding single lookup. let mut out = vec![LookupResult::DestinationMiss; inputs.len()]; dpdk.lookup_batch(&inputs, &mut out); @@ -288,12 +326,17 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { }); eprintln!( - "coverage: {} routes, {} source misses, {} destination misses", + "coverage: {} routes ({} revalidated), {} source misses, {} destination misses", ROUTES.load(Ordering::Relaxed), + REVALIDATED_ROUTES.load(Ordering::Relaxed), SOURCE_MISSES.load(Ordering::Relaxed), DESTINATION_MISSES.load(Ordering::Relaxed), ); assert!(ROUTES.load(Ordering::Relaxed) >= 1, "no full routes at all"); + assert!( + REVALIDATED_ROUTES.load(Ordering::Relaxed) >= 1, + "no route resolved through revalidation information", + ); assert!( SOURCE_MISSES.load(Ordering::Relaxed) >= 4, "too few source misses" @@ -305,10 +348,11 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { } /// The batched lookup equals the single lookup, slot for slot, on the reference backend. The -/// batch path's own logic -- the v4/v6 partition, `MAX_BATCH` chunking, the stage-1-hit gather -/// and the scatter back through saved indices -- is backend-generic, so this EAL-free variant -/// exercises it with far more iterations than the rte_acl differential can afford. 40 probes -/// force multi-chunk batches; probe specs freely mix IP versions and version-mismatched pairs. +/// batch path's own logic -- the v4/v6 partition, `MAX_BATCH` chunking, the stage-1-hit gather, +/// the revalidation re-lookups and the scatter back through saved indices -- is backend-generic, +/// so this EAL-free variant exercises it with far more iterations than the rte_acl differential +/// can afford. 40 probes force multi-chunk batches; probe specs freely mix IP versions and +/// version-mismatched pairs, and the overlay's derived probes bring the revalidated routes. #[test] fn batched_lookup_matches_single_lookup() { bolero::check!() @@ -318,37 +362,20 @@ fn batched_lookup_matches_single_lookup() { let tables = FlowFilterContext::build(&built.overlay, Backend::Reference) .expect("reference build"); - let probes: Vec = probe_specs - .iter() - .map(|p| p.resolve(built.blocks)) - .collect(); - let inputs: Vec = probes + let probes: Vec = built + .routing_probes .iter() - .map(|p| LookupInput { - src_vpcd: p.src_vpcd, - dst_vpcd: None, - src_ip: p.src_ip, - dst_ip: p.dst_ip, - proto: p.proto, - ports: p.ports, - nat_mode: None, - }) + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))) .collect(); + let inputs: Vec = probes.iter().map(lookup_input).collect(); let mut out = vec![LookupResult::DestinationMiss; inputs.len()]; tables.lookup_batch(&inputs, &mut out); for (i, probe) in probes.iter().enumerate() { assert_eq!( out[i], - tables.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ), + lookup(&tables, probe), "batch slot {i} != single lookup for {probe:?}\nspec: {overlay_spec:?}", ); } @@ -365,18 +392,14 @@ fn reference_lookup_matches_config_oracle() { let built = overlay_spec.build(); let tables = FlowFilterContext::build(&built.overlay, Backend::Reference).expect("reference build"); - for probe_spec in probe_specs { - let probe = probe_spec.resolve(built.blocks); + let probes = built + .routing_probes + .iter() + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))); + for probe in probes { assert_eq!( - tables.lookup( - probe.src_vpcd, - None, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - None, - ), + lookup(&tables, &probe), oracle_lookup(&built.overlay, &probe), "reference tables disagree with the config oracle on {probe:?}\nspec: {overlay_spec:?}", ); diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index 67ee4c193b..fb3fdfaefb 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -18,13 +18,18 @@ //! longest-prefix-match (encoded in the rule priority, see [`rule_priority`]) //! handles it uniformly. //! -//! Masquerade destinations are kept in the remote tables even though they cannot -//! accept new connections: their [`Verdict`] marks reply traffic on established -//! masquerade flows as distinguishable from a destination no peering covers, and -//! the NF gates them on flow state. Port-forwarding sources stay out of the local -//! tables (a covering expose must answer for connection initiation), so a stage-2 -//! miss is reported distinctly (see [`LookupResult`]) for the NF to resolve -//! against flow state. +//! Masquerade destinations cannot accept new connections and port-forwarding +//! sources cannot initiate them, so neither may answer a lookup on its own. Their +//! rules are still in the tables, keyed on the revalidation information an +//! outdated flow supplies: the destination VPC for the former, the source NAT mode +//! for the latter. Each stage therefore asks what the flow vouches for first, and +//! runs a second, ungated pass over the misses. A packet on an established flow +//! keeps the peering and the NAT mode that flow was built on, even where an expose +//! covers the same address ungated; the ungated pass is what answers for forward +//! traffic, which carries no flow information of its own. +//! A stage-2 miss is still reported distinctly (see [`LookupResult`]): a packet +//! whose flow the NF, not the tables, has to resolve reaches it without any +//! revalidation information. use crate::{NatMode, NatRequirement}; use acl::dpdk::dyn_table::predicate_to_chunks; @@ -460,8 +465,9 @@ fn emit_local( nat_mode: NatMode, action: NatMode, ) { - // Port-forwarding sources are never emitted into the local tables, so the tie-break bit is - // always clear here; local rules keep pure prefix-length ordering. + // Port-forwarding sources are the only local rules a masquerade rule can overlap, and the + // "nat_mode" key already keeps the two apart, so the tie-break bit is always clear here; + // local rules keep pure prefix-length ordering. let priority = rule_priority(ip_range, false); match ip_range { Prefix::IPV4(prefix) => { @@ -524,10 +530,10 @@ impl RuleSet { }; // Stage 1: peer's public prefixes -> Verdict{dst VPC, dst NAT}. Masquerade - // destinations cannot receive connections, but their rules stay in the table: - // a masquerade Verdict lets the NF tell reply traffic on an established - // masquerade flow apart from a destination no peering covers (which must drop). - // The NF only accepts a masquerade Verdict when the packet rides such a flow. + // destinations cannot receive connections, so their rules are keyed on the peer + // VNI: only a lookup revalidating against that very VPC -- reply traffic on an + // established masquerade flow -- reaches them. Everything else is keyed on 0, + // the "no revalidation" value, which is what an ordinary lookup asks with. for expose in peering.remote().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = Verdict { @@ -569,7 +575,9 @@ impl RuleSet { } // Stage 2: source's private prefixes -> source NAT mode. Port-forwarding sources - // cannot initiate connections, so they are excluded here. + // cannot initiate connections, so, symmetrically, their rules are keyed on the + // NAT mode they require: only a lookup revalidating against port forwarding + // reaches them. for expose in peering.local().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = NatRequirement::from_expose(expose); diff --git a/flow-filter/src/fuzz_gen.rs b/flow-filter/src/fuzz_gen.rs index 43b4bffdd5..3569b33191 100644 --- a/flow-filter/src/fuzz_gen.rs +++ b/flow-filter/src/fuzz_gen.rs @@ -23,6 +23,7 @@ #![cfg(test)] +use crate::{NatMode, NatRequirement}; use bolero::TypeGenerator; use config::external::overlay::vpc::{Vpc, VpcTable}; use config::external::overlay::vpcpeering::{VpcExpose, VpcManifest, VpcPeering, VpcPeeringTable}; @@ -135,6 +136,37 @@ impl ExposeSpec { ExposeSpec::PortForwarding(_) | ExposeSpec::PortFwProtoPair => None, } } + + /// Whether destinations of this expose are masquerade destinations: they cannot receive + /// connections, so the tables answer for them only when the lookup carries the destination VPC + /// an outdated flow revalidates against. + fn dest_needs_revalidation(self) -> bool { + matches!( + self, + ExposeSpec::Masquerade + | ExposeSpec::MasqueradeNestingPortFw(_) + | ExposeSpec::MasqueradeSameLenPortFw(_) + ) + } + + /// The port-forwarding exposes this spec contributes, as (protocol, host byte of an address + /// the private side covers). A port-forwarding source cannot initiate a connection, so the + /// tables answer for it only when the lookup carries the port-forwarding NAT mode an outdated + /// flow revalidates against -- and then in preference to any masquerade expose covering the + /// same address, which is why the specs that overlap the two are here too. + fn port_fw_sources(self) -> [Option<(FwProto, u8)>; 2] { + match self { + ExposeSpec::PortForwarding(proto) | ExposeSpec::MasqueradeNestingPortFw(proto) => { + [Some((proto, FW_HOST)), None] + } + // The port-forwarded prefix is the whole block, so any host of it will do. + ExposeSpec::MasqueradeSameLenPortFw(proto) => [Some((proto, 1)), None], + ExposeSpec::PortFwProtoPair => { + [Some((FwProto::Tcp, FW_HOST)), Some((FwProto::Udp, FW_HOST))] + } + ExposeSpec::Plain | ExposeSpec::StaticNat | ExposeSpec::Masquerade => [None, None], + } + } } /// An exclusion applied to an expose's private and public blocks. @@ -331,7 +363,7 @@ impl OverlaySpec { let remote = build_manifest(&vpc_name(b), &peering.remote, peering.v6, &mut blocks); derive_routing_probes( &mut routing_probes, - VNIS[a], + (VNIS[a], VNIS[b]), peering.v6, (local_base, &peering.local), (remote_base, &peering.remote), @@ -361,43 +393,91 @@ impl OverlaySpec { /// 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`. +/// Port-forwarding destinations target [`FW_HOST`] and [`FW_PUBLIC_PORTS`], port-forwarding +/// sources the host their expose covers and [`FW_PRIVATE_PORTS`]. Other endpoints use host `.1`. +/// +/// Masquerade destinations and port-forwarding sources have no rule a plain lookup can reach: the +/// tables gate them on the revalidation information an outdated flow supplies, so their probes +/// carry it. Both stages ask the revalidated question first, so such a probe routes through the +/// gated rule even where a catch-all or a masquerade expose covers the same address. fn derive_routing_probes( out: &mut Vec, - src_vni: u32, + (src_vni, dst_vni): (u32, u32), v6: bool, (local_base, local): (u8, &ManifestSpec), (remote_base, remote): (u8, &ManifestSpec), ) { let src_vpcd = VpcDiscriminant::from_vni(Vni::new_checked(src_vni).unwrap()); + let dst_vpcd = VpcDiscriminant::from_vni(Vni::new_checked(dst_vni).unwrap()); + + // (address, port, protocol, revalidated source NAT mode) of every source the peering routes. + // A protocol of None leaves the choice to the destination. + let mut sources: Vec<(IpAddr, u16, Option, NatMode)> = Vec::new(); for (li, lspec) in local.expose_specs().enumerate() { - if !lspec.source_capable() { - continue; + let block = local_base + li as u8; + if lspec.source_capable() { + sources.push((block_addr(block, 1, false, v6), 1, None, None)); } - let src_ip = block_addr(local_base + li as u8, 1, false, v6); - for (ri, rspec) in remote.expose_specs().enumerate() { - 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)), - }); - } + for (proto, host) in lspec.port_fw_sources().into_iter().flatten() { + sources.push(( + block_addr(block, host, false, v6), + FW_PRIVATE_PORTS.0, + Some(proto), + Some(NatRequirement::PortForwarding), + )); } } + + // (address, port, protocol, revalidated destination VPC) of every destination the peering + // routes. + let mut destinations: Vec<(IpAddr, u16, Option, Option)> = Vec::new(); + for (ri, rspec) in remote.expose_specs().enumerate() { + let block = remote_base + ri as u8; + if let Some(dst_public) = rspec.dest_public_space() { + let revalidated = rspec.dest_needs_revalidation().then_some(dst_vpcd); + destinations.push((block_addr(block, 1, dst_public, v6), 1, None, revalidated)); + } + for proto in rspec.portfw_protos() { + destinations.push(( + block_addr(block, FW_HOST, true, v6), + FW_PUBLIC_PORTS.0, + Some(proto), + None, + )); + } + } + + for &(src_ip, src_port, src_proto, nat_mode) in &sources { + for &(dst_ip, dst_port, dst_proto, revalidated_dst) in &destinations { + let Some(proto) = pair_proto(src_proto, dst_proto) else { + continue; + }; + out.push(Probe { + src_vpcd, + dst_vpcd: revalidated_dst, + src_ip, + dst_ip, + proto, + ports: Some((src_port, dst_port)), + nat_mode, + }); + } + } +} + +/// The protocol a probe pairing these two endpoints carries, or `None` if each end constrains it +/// to a protocol the other rejects. An end that constrains nothing follows the other one, and a +/// pair that constrains nothing uses TCP. +fn pair_proto(src: Option, dst: Option) -> Option { + match (src, dst) { + (None, None) => Some(NextHeader::TCP), + (Some(proto), None) | (None, Some(proto)) => Some(proto.probe_next_header()), + (Some(FwProto::Any), Some(proto)) | (Some(proto), Some(FwProto::Any)) => { + Some(proto.probe_next_header()) + } + (Some(src), Some(dst)) if src == dst => Some(src.probe_next_header()), + (Some(_), Some(_)) => None, + } } fn vpc_name(index: usize) -> String { @@ -590,6 +670,24 @@ impl ProbeProto { } } +/// The NAT mode an outdated flow revalidates a packet's source against. +#[derive(Debug, Clone, Copy, TypeGenerator)] +pub(crate) enum NatSel { + Static, + Masquerade, + PortForwarding, +} + +impl NatSel { + fn requirement(self) -> NatRequirement { + match self { + NatSel::Static => NatRequirement::Static, + NatSel::Masquerade => NatRequirement::Masquerade, + NatSel::PortForwarding => NatRequirement::PortForwarding, + } + } +} + /// One packet's lookup question, in spec form. Block selectors are reduced modulo the built /// overlay's block count so probes usually land inside some generated prefix; hits require the /// blocks to pair up with the right peering, misses come for free. @@ -608,28 +706,42 @@ pub(crate) struct ProbeSpec { proto: ProbeProto, sport: PortSel, dport: PortSel, + /// Revalidation information, as an outdated flow would supply it: the destination VPC the flow + /// was routed to, and the NAT mode it carries. Both are drawn freely (any VPC, any mode, + /// neither), so the lookups see combinations no flow would produce as well. + revalidate_dst: Option, + revalidate_nat: Option, } /// A resolved probe: the arguments of one route lookup. #[derive(Debug, Clone, Copy)] pub(crate) struct Probe { pub(crate) src_vpcd: VpcDiscriminant, + pub(crate) dst_vpcd: Option, pub(crate) src_ip: IpAddr, pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, + pub(crate) nat_mode: NatMode, +} + +/// The VPC discriminant a `u8` selector draws: one of the generated VPCs, or the bogus one. +fn vpcd_from_sel(sel: u8) -> VpcDiscriminant { + let vni = match sel as usize % (VNIS.len() + 1) { + i if i < VNIS.len() => VNIS[i], + _ => BOGUS_VNI, + }; + VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap()) } impl ProbeSpec { pub(crate) fn resolve(&self, blocks: u8) -> Probe { let nblocks = blocks.max(1); - let vni = match self.vni_sel as usize % (VNIS.len() + 1) { - i if i < VNIS.len() => VNIS[i], - _ => BOGUS_VNI, - }; let dst_v6 = self.v6 ^ self.cross_version; Probe { - src_vpcd: VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap()), + src_vpcd: vpcd_from_sel(self.vni_sel), + dst_vpcd: self.revalidate_dst.map(vpcd_from_sel), + nat_mode: self.revalidate_nat.map(NatSel::requirement), src_ip: block_addr( self.src_block % nblocks, self.src_host, diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index 152cb18d4e..fed14db86d 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -1112,6 +1112,9 @@ fn probe_from_packet(pkt: &Packet, src_vpcd: VpcDiscriminant) -> Opt let net = pkt.try_ip()?; Some(Probe { src_vpcd, + // These packets belong to no flow, so they don't need flow revalidation info. + dst_vpcd: None, + nat_mode: None, src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -1145,6 +1148,11 @@ fn probe_packet(probe: &Probe) -> Option<(Packet, Probe)> { use net::ip::NextHeader; let mut probe = *probe; + // The NF sees these packets without a flow, so the revalidation information a derived probe + // carries never reaches the lookup: the oracle must not try to lookup for revalidation info + // that the NF cannot see, so we clear revalidation info. + probe.dst_vpcd = None; + probe.nat_mode = None; if let Some((sport, dport)) = probe.ports.as_mut() { *sport = (*sport).max(1); *dport = (*dport).max(1); From 9c7e9b6481403e1c67039990d06dfb3f21c5e255 Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 16:17:14 +0100 Subject: [PATCH 4/8] test(flow-filter): Add unit tests for flow revalidation + prefix overlap Make sure flow revalidation when flow info is outdated (with regards to flow-filter's genid) behaves as we expect, in particular for the two following (distinct) cases: - We have overlapping, masqueraded prefixes exposed by multiples VPCs to a given VPC - We have overlapping prefixes for port-forwarding and masquerading, within a given manifest. For both port-forwarding and masquerade in this setting, check the behaviour for the initial packet, follow-up replies in both directions; then update the genid and make sure that we still resolve for the different packets as expected. Signed-off-by: Quentin Monnet --- flow-filter/src/tests.rs | 276 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 276 insertions(+) diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index fed14db86d..e9a360bd61 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -675,6 +675,282 @@ fn stateful_flow_does_not_survive_peering_removal() { assert_eq!(flow.status(), FlowStatus::Cancelled); } +// ------------------------------------------------------------------------------------------------- +// Config update and flow re-validation in the case of overlaps + +#[test] +fn revalidation_works_in_case_of_remote_masquerade_overlap() { + let ctx = context( + &[("vpc1", 100), ("vpc2", 200), ("vpc3", 300)], + // vpc2 and vpc3 both expose the same masqueraded prefixes towards the same vpc1 prefix + vec![ + peering( + "vpc1-to-vpc2", + ("vpc1", vec![expose("1.0.0.0/24")]), + ("vpc2", vec![expose_masquerade("2.0.0.0/24", "10.0.0.0/24")]), + ), + peering( + "vpc1-to-vpc3", + ("vpc1", vec![expose("1.0.0.0/24")]), + ("vpc3", vec![expose_masquerade("2.0.0.0/24", "10.0.0.0/24")]), + ), + ], + ); + let (mut flow_filter, writer) = make_flow_filter(ctx); + + // Initial packet from vpc2 to vpc1 (no flow info) passes + let p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + + // Reply from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Request from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Bump flow-filter genid + set_genid(&mut flow_filter, 5); + + // Reply from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Reply from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Rmove peering, bump genid again + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, 6); + + // Reply from vpc1 to vpc2 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + +#[test] +fn revalidation_works_in_case_of_local_masquerade_portforwarding_overlap() { + let ctx = context( + &[("vpc1", 100), ("vpc2", 200)], + // vpc1 uses overlapping prefixes for masquerade and port-forwarding + vec![peering( + "vpc1-to-vpc2", + ( + "vpc1", + vec![ + expose_port_forwarding( + "1.0.0.0/24", + (2000, 3000), + "10.0.0.0/24", + (5000, 6000), + Some(L4Protocol::Tcp), + ), + expose_masquerade("1.0.0.0/25", "10.0.0.0/25"), + ], + ), + ("vpc2", vec![expose("2.0.0.0/24")]), + )], + ); + let (mut flow_filter, writer) = make_flow_filter(ctx); + + // Port-forwarding: Initial packet from vpc2 to vpc1 (no flow info) passes + let p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + + // Port-forwarding: Reply from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Port-forwarding: Request from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Masquerade: Initial packet from vpc1 to vpc2 (no flow info) passes + let p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + + // Masquerade: Reply from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Request from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Bump flow-filter genid + set_genid(&mut flow_filter, 5); + + // ------ + + // Port-forwarding: Reply from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Port-forwarding: Request from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Masquerade: Reply from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Request from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Rmove peering, bump genid again + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, 6); + + // Port-forwarding: Reply from vpc1 to vpc2 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Reply from vpc2 to vpc1 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + // ------------------------------------------------------------------------------------------------- // Stateful flows: flow-key attachment for the {masquerade|port-forwarding} + static-NAT combination From c19f1f8d595d22a6b3dad88d9e4a62e2e8bea3c4 Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 17:51:37 +0100 Subject: [PATCH 5/8] feat(flow-filter): Use Option rather than u32 in remote table We recently added the destination VNI as part of the key in the remote table for the flow-filter lookup, to use only in the case when we want to revalidate the flow for reply masqueraded traffic with outdated flow information. Rather than using a u32 (and 0 when we don't want to use the VNI), use an Option, wrapped withing a GateVni type so we can implement FixedSize for it. No functional change. Co-authored-by: Claude Opus 5 (1M context) Signed-off-by: Quentin Monnet --- flow-filter/src/context/display.rs | 11 ++++- flow-filter/src/context/tables.rs | 67 ++++++++++++++++++++---------- flow-filter/src/context/tests.rs | 2 +- 3 files changed, 55 insertions(+), 25 deletions(-) diff --git a/flow-filter/src/context/display.rs b/flow-filter/src/context/display.rs index b9561c9bc6..0daf805afa 100644 --- a/flow-filter/src/context/display.rs +++ b/flow-filter/src/context/display.rs @@ -6,7 +6,7 @@ //! Tables retain typed rules for backend-independent display. Each field's type controls its //! formatting, keeping values coupled to their key fields. -use super::tables::FlowFilterContext; +use super::tables::{FlowFilterContext, GateVni}; impl crate::NatRequirement { fn label(self) -> &'static str { @@ -24,6 +24,15 @@ impl std::fmt::Display for crate::NatRequirement { } } +impl std::fmt::Display for GateVni { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self.0 { + Some(vni) => write!(f, "{vni}"), + None => f.write_str("-"), + } + } +} + // ------------------------------------------------------------------------------------------------- // Rendering: one section per table, each rule on a line, in match order. diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index fb3fdfaefb..8e3f5a6bea 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -102,7 +102,7 @@ pub(super) struct Verdict { /// concrete; every other field is carried verbatim from the [`LookupInput`]). struct Query { src_vni: Vni, - dst_vni: u32, + dst_vni: GateVni, proto: NextHeader, src_ip: I, dst_ip: I, @@ -128,6 +128,32 @@ fn key_vni(vpcd: VpcDiscriminant) -> Vni { } } +/// The destination VPC a stage-1 rule is gated on, or `None` for a rule an ungated lookup reaches. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(transparent)] +pub(super) struct GateVni(pub(crate) Option); + +impl GateVni { + const UNGATED: Self = Self(None); + + fn is_gated(self) -> bool { + self.0.is_some() + } +} + +impl From> for GateVni { + fn from(vni: Option) -> Self { + Self(vni) + } +} + +impl FixedSize for GateVni { + const SIZE: usize = Vni::SIZE; + fn write_be(&self, out: &mut [u8]) { + self.0.map_or(0, Vni::as_u32).write_be(out); + } +} + // ------------------------------------------------------------------------------------------------- // Keys. // @@ -146,7 +172,7 @@ pub(super) struct RemoteKey { src_vni: Vni, #[exact] #[cli(column_name = "dst-vni")] - dst_vni: u32, + dst_vni: GateVni, #[prefix] #[cli(column_name = "destination")] dst_ip: I, @@ -408,7 +434,7 @@ fn emit_remote( v4: &mut Vec, Verdict>>, v6: &mut Vec, Verdict>>, src_vni: Vni, - dst_vni: u32, + dst_vni: GateVni, ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, @@ -530,21 +556,17 @@ impl RuleSet { }; // Stage 1: peer's public prefixes -> Verdict{dst VPC, dst NAT}. Masquerade - // destinations cannot receive connections, so their rules are keyed on the peer + // destinations cannot receive connections, so their rules are gated on the peer // VNI: only a lookup revalidating against that very VPC -- reply traffic on an - // established masquerade flow -- reaches them. Everything else is keyed on 0, - // the "no revalidation" value, which is what an ordinary lookup asks with. + // established masquerade flow -- reaches them. Everything else is ungated, which + // is what an ordinary lookup asks with. for expose in peering.remote().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = Verdict { nat_mode: NatRequirement::from_expose(expose), dst_vpcd: remote_vpcd, }; - let dst_vni = if expose.has_masquerade() { - remote_vni.as_u32() - } else { - 0 - }; + let dst_vni = GateVni::from(expose.has_masquerade().then_some(remote_vni)); for prefix in expose.public_ips() { emit_remote( &mut rules.remote_v4, @@ -563,7 +585,7 @@ impl RuleSet { &mut rules.remote_v4, &mut rules.remote_v6, src_vni, - 0, + GateVni::UNGATED, default_ip(), PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), @@ -682,7 +704,7 @@ impl FlowFilterContext { nat_mode: NatMode, ) -> LookupResult { let src_vni = key_vni(src_vpcd); - let dst_vni = dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); + let dst_vni = GateVni::from(dst_vpcd.map(key_vni)); let (src_port, dst_port) = ports.unzip(); let src_port = src_port.unwrap_or(0); let dst_port = dst_port.unwrap_or(0); @@ -699,11 +721,11 @@ impl FlowFilterContext { }) { v } else { - if dst_vni != 0 + if dst_vni.is_gated() && let Some(v) = self.remote_v4.lookup(&RemoteKey { proto, src_vni, - dst_vni: 0, + dst_vni: GateVni::UNGATED, dst_ip, dst_port, }) @@ -753,11 +775,11 @@ impl FlowFilterContext { }) { v } else { - if dst_vni != 0 + if dst_vni.is_gated() && let Some(v) = self.remote_v6.lookup(&RemoteKey { proto, src_vni, - dst_vni: 0, + dst_vni: GateVni::UNGATED, dst_ip, dst_port, }) @@ -823,7 +845,7 @@ impl FlowFilterContext { out[i] = LookupResult::DestinationMiss; let proto = input.proto; let src_vni = key_vni(input.src_vpcd); - let dst_vni = input.dst_vpcd.map(|d| key_vni(d).as_u32()).unwrap_or(0); + let dst_vni = GateVni::from(input.dst_vpcd.map(key_vni)); let (src_port, dst_port) = input.ports.unwrap_or((0, 0)); let nat_mode = NatRequirement::convert_option(input.nat_mode); match (input.src_ip, input.dst_ip) { @@ -893,18 +915,17 @@ fn lookup_versioned( // Reply traffic for masqueraded flows use the destination VNI as part of the key; this is // to avoid conflicting entries if there are several VPCs exposing overlapping, masqueraded // prefixes to a given VPC. If we have a destination VNI set here, we may be trying to - // re-validated a reply packet for a masqueraded flow (we're not sure of the direction, - // hence the first attempt with the destination VNI set to 0 above). Try again after setting - // the destination VNI. + // re-validate a reply packet for a masqueraded flow (we're not sure of the direction, hence + // the first attempt with the destination VNI). Try again, without the destination VNI. let mut reval_positions = Vec::new(); let mut reval_keys = Vec::new(); for (pos, (query, verdict)) in q_chunk.iter().zip(verdicts.iter_mut()).enumerate() { - if verdict.is_none() && query.dst_vni != 0 { + if verdict.is_none() && query.dst_vni.is_gated() { reval_positions.push(pos); reval_keys.push(RemoteKey { proto: query.proto, src_vni: query.src_vni, - dst_vni: 0, + dst_vni: GateVni::UNGATED, dst_ip: query.dst_ip, dst_port: query.dst_port, }); diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index cab533a58b..e6006ca12c 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -923,7 +923,7 @@ fn display_is_identical_across_backends() { "[0]", "TCP", "100", - "0", + "-", "80.0.0.5/32", "2222", "|", From 22fd66e8171d9903bb0fa3b6c1a3696e6fa2c770 Mon Sep 17 00:00:00 2001 From: Quentin Monnet Date: Thu, 6 Aug 2026 18:04:27 +0100 Subject: [PATCH 6/8] feat(flow-filter): Use SourceGate instead of u8 in local table We used to represent the NAT mode in the local table key with a u8; use a dedicated "gate" instead (a flag to turn on and off the lookups for entries for reply traffic with port forwarding for revalidating flows with outdated information, in the local-side context table), similarly to what we did with GateVni for the remote table. Co-authored-by: Claude Opus 5 (1M context) Signed-off-by: Quentin Monnet --- flow-filter/src/context/display.rs | 11 +++- flow-filter/src/context/fuzz.rs | 38 ++++++------- flow-filter/src/context/mod.rs | 2 +- flow-filter/src/context/tables.rs | 87 +++++++++++++++++++----------- flow-filter/src/context/tests.rs | 45 +++++++++++----- flow-filter/src/fuzz_gen.rs | 51 ++++++++---------- flow-filter/src/lib.rs | 39 +++++--------- flow-filter/src/tests.rs | 5 +- 8 files changed, 153 insertions(+), 125 deletions(-) diff --git a/flow-filter/src/context/display.rs b/flow-filter/src/context/display.rs index 0daf805afa..267bc3d74d 100644 --- a/flow-filter/src/context/display.rs +++ b/flow-filter/src/context/display.rs @@ -6,7 +6,7 @@ //! Tables retain typed rules for backend-independent display. Each field's type controls its //! formatting, keeping values coupled to their key fields. -use super::tables::{FlowFilterContext, GateVni}; +use super::tables::{FlowFilterContext, GateVni, SourceGate}; impl crate::NatRequirement { fn label(self) -> &'static str { @@ -33,6 +33,15 @@ impl std::fmt::Display for GateVni { } } +impl std::fmt::Display for SourceGate { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SourceGate::Ungated => f.write_str("-"), + SourceGate::PortFwdReply => f.write_str("pfwd"), + } + } +} + // ------------------------------------------------------------------------------------------------- // Rendering: one section per table, each rule on a line, in match order. diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index ca56054397..8f920d2ed6 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -11,9 +11,9 @@ #![cfg(test)] -use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult}; +use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult, SourceGate}; +use crate::NatRequirement; use crate::fuzz_gen::{OverlaySpec, Probe, ProbeSpec, bogus_vpcd}; -use crate::{NatMode, NatRequirement}; use concurrency::sync::LazyLock; use concurrency::sync::atomic::{AtomicU64, Ordering}; use config::external::overlay::ValidatedOverlay; @@ -116,23 +116,21 @@ fn oracle_stage1( /// Stage 2: the source against the resolved peering's private prefixes. /// -/// `revalidated` is the source NAT mode an outdated flow vouches for. With it, only the -/// port-forwarding exposes of that very mode answer -- they cannot initiate connections, so -/// nothing but a flow that already used them can reach them. Without it, only exposes that can -/// initiate do (a default expose acts as a /0 of the peering's IP version). +/// `gate` is what the lookup asks about. Gated on port forwarding, only the port-forwarding +/// exposes answer -- they cannot initiate connections, so nothing but a flow that already used +/// them can reach them. Ungated, only exposes that can initiate do (a default expose acts as a /0 +/// of the peering's IP version). fn oracle_stage2( peering: &ValidatedPeering, probe: &Probe, sport: u16, - revalidated: NatMode, + gate: SourceGate, ) -> Option> { let mut src_nat: Option<(Precedence, Option)> = None; for expose in peering.local().valexp() { - let matchable = match revalidated { - Some(mode) => { - expose.has_port_forwarding() && NatRequirement::from_expose(expose) == Some(mode) - } - None => expose.can_init_connection(), + let matchable = match gate { + SourceGate::PortFwdReply => expose.has_port_forwarding(), + SourceGate::Ungated => expose.can_init_connection(), }; if !matchable || !proto_allows(expose.nat_proto(), probe.proto) { continue; @@ -147,7 +145,7 @@ fn oracle_stage2( } } } - if revalidated.is_none() + if !gate.is_gated() && peering.local().has_default_expose() && probe.src_ip.is_ipv4() == peering.is_v4() { @@ -189,10 +187,8 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup .iter() .find(|p| VpcDiscriminant::from_vni(p.remote_vni()) == dst_vpcd) .unwrap_or_else(|| unreachable!("stage 1 hit implies a peering to the verdict VPC")); - let src_nat = probe - .nat_mode - .and_then(|mode| oracle_stage2(peering, probe, sport, Some(mode))) - .or_else(|| oracle_stage2(peering, probe, sport, None)); + let src_nat = oracle_stage2(peering, probe, sport, probe.gate) + .or_else(|| oracle_stage2(peering, probe, sport, SourceGate::Ungated)); match src_nat { Some(src_nat) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), None => LookupResult::SourceMiss(dst_vpcd), @@ -211,7 +207,7 @@ fn lookup(tables: &FlowFilterContext, probe: &Probe) -> LookupResult { probe.dst_ip, probe.proto, probe.ports, - probe.nat_mode, + probe.gate, ) } @@ -224,7 +220,7 @@ fn lookup_input(probe: &Probe) -> LookupInput { dst_ip: probe.dst_ip, proto: probe.proto, ports: probe.ports, - nat_mode: probe.nat_mode, + gate: probe.gate, } } @@ -262,7 +258,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { matches!(lookup(&reference, probe), LookupResult::Route(_)), "derived routing probe did not route: {probe:?}\nspec: {overlay_spec:?}", ); - if probe.dst_vpcd.is_some() || probe.nat_mode.is_some() { + if probe.dst_vpcd.is_some() || probe.gate.is_gated() { REVALIDATED_ROUTES.fetch_add(1, Ordering::Relaxed); } } @@ -314,7 +310,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { dst_ip: "10.0.0.99".parse().unwrap(), proto: NextHeader::TCP, ports: Some((1, 2)), - nat_mode: None, + gate: SourceGate::Ungated, }) .collect(); let mut out = vec![LookupResult::DestinationMiss; all_miss.len()]; diff --git a/flow-filter/src/context/mod.rs b/flow-filter/src/context/mod.rs index cd703a8403..4d1cb245fe 100644 --- a/flow-filter/src/context/mod.rs +++ b/flow-filter/src/context/mod.rs @@ -17,7 +17,7 @@ mod tests; pub use tables::FlowFilterContext; use tables::PRODUCTION_BACKEND; -pub(crate) use tables::{LookupInput, LookupResult}; +pub(crate) use tables::{LookupInput, LookupResult, SourceGate}; impl TryFrom<&ValidatedOverlay> for FlowFilterContext { type Error = ConfigError; diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index 8e3f5a6bea..cba0536f90 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -88,7 +88,7 @@ pub(crate) struct LookupInput { pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, - pub(crate) nat_mode: NatMode, + pub(crate) gate: SourceGate, } /// Result of a stage-1 (remote/destination) match. @@ -108,7 +108,7 @@ struct Query { dst_ip: I, src_port: u16, dst_port: u16, - nat_mode: u8, + gate: SourceGate, } /// Lower a config L4 protocol to a bitmask predicate: a specific protocol matches exactly (every @@ -154,6 +154,34 @@ impl FixedSize for GateVni { } } +/// A flag to gate what a stage-2 entry answers for: +/// +/// - "revalidation of reply traffic associated with port-forwarding, with outdated flow info" +/// - everything else ([`Ungated`]) +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) enum SourceGate { + #[default] + Ungated, + PortFwdReply, +} + +impl SourceGate { + pub(crate) fn is_gated(self) -> bool { + self != Self::Ungated + } +} + +impl FixedSize for SourceGate { + const SIZE: usize = u8::SIZE; + fn write_be(&self, out: &mut [u8]) { + match self { + Self::Ungated => 0u8, + Self::PortFwdReply => 1u8, + } + .write_be(out); + } +} + // ------------------------------------------------------------------------------------------------- // Keys. // @@ -199,8 +227,8 @@ pub(super) struct LocalKey { #[cli(column_name = "src-port")] src_port: u16, #[exact] - #[cli(column_name = "nat-mode")] - nat_mode: u8, + #[cli(column_name = "gate")] + gate: SourceGate, } // ------------------------------------------------------------------------------------------------- @@ -488,7 +516,7 @@ fn emit_local( ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, - nat_mode: NatMode, + gate: SourceGate, action: NatMode, ) { // Port-forwarding sources are the only local rules a masquerade rule can overlap, and the @@ -503,7 +531,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, - nat_mode: ExactSpec::new(NatRequirement::convert_option(nat_mode)), + gate: ExactSpec::new(gate), }; v4.push(NeutralRule { priority, @@ -519,7 +547,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, - nat_mode: ExactSpec::new(NatRequirement::convert_option(nat_mode)), + gate: ExactSpec::new(gate), }; v6.push(NeutralRule { priority, @@ -597,16 +625,16 @@ impl RuleSet { } // Stage 2: source's private prefixes -> source NAT mode. Port-forwarding sources - // cannot initiate connections, so, symmetrically, their rules are keyed on the + // cannot initiate connections, so, symmetrically, their rules are gated on the // NAT mode they require: only a lookup revalidating against port forwarding // reaches them. for expose in peering.local().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = NatRequirement::from_expose(expose); - let nat_mode = if expose.has_port_forwarding() { - action + let gate = if expose.has_port_forwarding() { + SourceGate::PortFwdReply } else { - None + SourceGate::Ungated }; for prefix in expose.ips() { emit_local( @@ -617,7 +645,7 @@ impl RuleSet { prefix.prefix(), prefix.into(), proto, - nat_mode, + gate, action, ); } @@ -631,7 +659,7 @@ impl RuleSet { default_ip(), PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), - None, + SourceGate::Ungated, None, ); } @@ -701,14 +729,13 @@ impl FlowFilterContext { dst_ip: IpAddr, proto: NextHeader, ports: Option<(u16, u16)>, - nat_mode: NatMode, + gate: SourceGate, ) -> LookupResult { let src_vni = key_vni(src_vpcd); let dst_vni = GateVni::from(dst_vpcd.map(key_vni)); let (src_port, dst_port) = ports.unzip(); let src_port = src_port.unwrap_or(0); let dst_port = dst_port.unwrap_or(0); - let nat_mode = NatRequirement::convert_option(nat_mode); match (src_ip, dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { @@ -742,20 +769,20 @@ impl FlowFilterContext { dst_vni, src_ip, src_port, - nat_mode, + gate, }) { Some(src_nat_mode) => { LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) } None => { - if nat_mode != 0 + if gate.is_gated() && let Some(src_nat_mode) = self.local_v4.lookup(&LocalKey { proto, src_vni, dst_vni, src_ip, src_port, - nat_mode: 0, + gate: SourceGate::Ungated, }) { LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) @@ -796,20 +823,20 @@ impl FlowFilterContext { dst_vni, src_ip, src_port, - nat_mode, + gate, }) { Some(src_nat_mode) => { LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) } None => { - if nat_mode != 0 + if gate.is_gated() && let Some(src_nat_mode) = self.local_v6.lookup(&LocalKey { proto, src_vni, dst_vni, src_ip, src_port, - nat_mode: 0, + gate: SourceGate::Ungated, }) { LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) @@ -847,7 +874,7 @@ impl FlowFilterContext { let src_vni = key_vni(input.src_vpcd); let dst_vni = GateVni::from(input.dst_vpcd.map(key_vni)); let (src_port, dst_port) = input.ports.unwrap_or((0, 0)); - let nat_mode = NatRequirement::convert_option(input.nat_mode); + let gate = input.gate; match (input.src_ip, input.dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { v4_idx.push(i); @@ -859,7 +886,7 @@ impl FlowFilterContext { dst_ip, src_port, dst_port, - nat_mode, + gate, }); } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { @@ -872,7 +899,7 @@ impl FlowFilterContext { dst_ip, src_port, dst_port, - nat_mode, + gate, }); } _ => { /* version mismatch: leave "out[i] = DestinationMiss" */ } @@ -951,7 +978,7 @@ fn lookup_versioned( dst_vni: key_vni(verdict.dst_vpcd), src_ip: q.src_ip, src_port: q.src_port, - nat_mode: q.nat_mode, + gate: q.gate, }); hit_pos.push(pos); } @@ -960,16 +987,16 @@ fn lookup_versioned( local.lookup_batch(&local_keys, &mut nat_modes); // Second pass: if nat_mode was set and we didn't find an entry for reply traffic associated - // with a port-forwarding flow, set nat_mode to 0 to see if we have an entry for forward - // traffic for port-forwarding (forward traffic entries do not have flow-info nat mode - // attached, or we couldn't use it to initiate new flows). + // with a port-forwarding flow, drop the gate to see if we have an entry for forward traffic + // for port-forwarding (forward traffic entries do not have flow-info nat mode attached, or + // we couldn't use it to initiate new flows). let mut reval_positions = Vec::new(); let mut reval_keys = Vec::new(); for (pos, (nat_mode, &q_pos)) in nat_modes.iter().zip(hit_pos.iter()).enumerate() { - if nat_mode.is_none() && q_chunk[q_pos].nat_mode != 0 { + if nat_mode.is_none() && q_chunk[q_pos].gate.is_gated() { reval_positions.push(pos); reval_keys.push(LocalKey { - nat_mode: 0, + gate: SourceGate::Ungated, ..local_keys[pos].clone() }); } diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index e6006ca12c..95d3495ad2 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -7,6 +7,7 @@ use super::LookupResult; use super::tables::RuleRow; +use super::tables::SourceGate; use crate::test_utils::*; use crate::{FlowFilterContext, NatMode, NatRequirement}; use lpm::prefix::L4Protocol; @@ -29,7 +30,7 @@ fn route( src_vpcd: VpcDiscriminant, headers: &Headers, ) -> Option { - match route_lookup(context, src_vpcd, None, None, headers) { + match route_lookup(context, src_vpcd, None, SourceGate::Ungated, headers) { LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { dst_vpcd, dst_nat, @@ -45,7 +46,7 @@ fn route_lookup( context: &FlowFilterContext, src_vpcd: VpcDiscriminant, dst_vpcd: Option, - nat_mode: NatMode, + gate: SourceGate, headers: &Headers, ) -> LookupResult { let net = headers.net().unwrap(); @@ -57,7 +58,7 @@ fn route_lookup( .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }); - context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports, nat_mode) + context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports, gate) } // ------------------------------------------------------------------------------------------------- @@ -315,7 +316,7 @@ fn dst_side_nat_modes() { &ctx, vpcd(100), None, - None, + SourceGate::Ungated, &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), ); assert_eq!(lookup_result, LookupResult::DestinationMiss); @@ -326,7 +327,7 @@ fn dst_side_nat_modes() { &ctx, vpcd(100), Some(vpcd(200)), - None, + SourceGate::Ungated, &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), ) else { panic!("masquerade destination resolves as a marker"); @@ -351,7 +352,7 @@ fn dst_side_nat_modes() { &ctx, vpcd(200), None, - None, + SourceGate::Ungated, &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), ); assert_eq!(lookup_result, LookupResult::SourceMiss(vpcd(100))); @@ -361,7 +362,7 @@ fn dst_side_nat_modes() { &ctx, vpcd(200), None, - Some(NatRequirement::PortForwarding), + SourceGate::PortFwdReply, &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), ) else { panic!("masquerade destination resolves as a marker"); @@ -522,7 +523,7 @@ fn port_forwarding_and_masquerade_overlap_resoves_as_expected() { &ctx, vpcd(100), None, - Some(NatRequirement::PortForwarding), + SourceGate::PortFwdReply, &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), ) else { panic!("source resolves via the port-forwarding expose"); @@ -535,7 +536,7 @@ fn port_forwarding_and_masquerade_overlap_resoves_as_expected() { &ctx, vpcd(100), None, - None, + SourceGate::Ungated, &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), ) else { panic!("source resolves via the masquerade expose"); @@ -795,8 +796,24 @@ fn reference_and_dpdk_backends_agree() { for &(vni, src_ip, dst_ip, proto, ports) in probes { let src_vpcd = vpcd(vni); assert_eq!( - reference.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports, None), - dpdk.lookup(src_vpcd, None, src_ip, dst_ip, proto, ports, None), + reference.lookup( + src_vpcd, + None, + src_ip, + dst_ip, + proto, + ports, + SourceGate::Ungated + ), + dpdk.lookup( + src_vpcd, + None, + src_ip, + dst_ip, + proto, + ports, + SourceGate::Ungated + ), "backends disagree on {src_ip} -> {dst_ip} ({proto:?}) from vni {vni}", ); } @@ -813,7 +830,7 @@ fn reference_and_dpdk_backends_agree() { dst_ip, proto, ports, - nat_mode: None, + gate: SourceGate::Ungated, }) .collect(); assert!(inputs.len() > 32, "want a multi-chunk batch"); @@ -832,7 +849,7 @@ fn reference_and_dpdk_backends_agree() { input.dst_ip, input.proto, input.ports, - input.nat_mode, + input.gate, ); assert_eq!(ref_out[i], single, "batched != single at index {i}"); } @@ -938,7 +955,7 @@ fn display_is_identical_across_backends() { assert_eq!( cells(&local_v4, "rank"), [ - "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "nat-mode", "|", "NAT" + "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "gate", "|", "NAT" ], "unexpected local heading row:\n{local_v4}" ); diff --git a/flow-filter/src/fuzz_gen.rs b/flow-filter/src/fuzz_gen.rs index 3569b33191..097d2480df 100644 --- a/flow-filter/src/fuzz_gen.rs +++ b/flow-filter/src/fuzz_gen.rs @@ -23,7 +23,7 @@ #![cfg(test)] -use crate::{NatMode, NatRequirement}; +use crate::context::SourceGate; use bolero::TypeGenerator; use config::external::overlay::vpc::{Vpc, VpcTable}; use config::external::overlay::vpcpeering::{VpcExpose, VpcManifest, VpcPeering, VpcPeeringTable}; @@ -412,18 +412,23 @@ fn derive_routing_probes( // (address, port, protocol, revalidated source NAT mode) of every source the peering routes. // A protocol of None leaves the choice to the destination. - let mut sources: Vec<(IpAddr, u16, Option, NatMode)> = Vec::new(); + let mut sources: Vec<(IpAddr, u16, Option, SourceGate)> = Vec::new(); for (li, lspec) in local.expose_specs().enumerate() { let block = local_base + li as u8; if lspec.source_capable() { - sources.push((block_addr(block, 1, false, v6), 1, None, None)); + sources.push(( + block_addr(block, 1, false, v6), + 1, + None, + SourceGate::Ungated, + )); } for (proto, host) in lspec.port_fw_sources().into_iter().flatten() { sources.push(( block_addr(block, host, false, v6), FW_PRIVATE_PORTS.0, Some(proto), - Some(NatRequirement::PortForwarding), + SourceGate::PortFwdReply, )); } } @@ -447,7 +452,7 @@ fn derive_routing_probes( } } - for &(src_ip, src_port, src_proto, nat_mode) in &sources { + for &(src_ip, src_port, src_proto, gate) in &sources { for &(dst_ip, dst_port, dst_proto, revalidated_dst) in &destinations { let Some(proto) = pair_proto(src_proto, dst_proto) else { continue; @@ -459,7 +464,7 @@ fn derive_routing_probes( dst_ip, proto, ports: Some((src_port, dst_port)), - nat_mode, + gate, }); } } @@ -670,24 +675,6 @@ impl ProbeProto { } } -/// The NAT mode an outdated flow revalidates a packet's source against. -#[derive(Debug, Clone, Copy, TypeGenerator)] -pub(crate) enum NatSel { - Static, - Masquerade, - PortForwarding, -} - -impl NatSel { - fn requirement(self) -> NatRequirement { - match self { - NatSel::Static => NatRequirement::Static, - NatSel::Masquerade => NatRequirement::Masquerade, - NatSel::PortForwarding => NatRequirement::PortForwarding, - } - } -} - /// One packet's lookup question, in spec form. Block selectors are reduced modulo the built /// overlay's block count so probes usually land inside some generated prefix; hits require the /// blocks to pair up with the right peering, misses come for free. @@ -706,11 +693,11 @@ pub(crate) struct ProbeSpec { proto: ProbeProto, sport: PortSel, dport: PortSel, - /// Revalidation information, as an outdated flow would supply it: the destination VPC the flow - /// was routed to, and the NAT mode it carries. Both are drawn freely (any VPC, any mode, - /// neither), so the lookups see combinations no flow would produce as well. + /// Revalidation information, as an outdated flow would supply it: the destination VPC the + /// flow was routed to, and whether it is a port-forwarding flow. Both are drawn freely (any + /// VPC, either gate, neither), so the lookups see combinations no flow would produce as well. revalidate_dst: Option, - revalidate_nat: Option, + revalidate_port_fw: bool, } /// A resolved probe: the arguments of one route lookup. @@ -722,7 +709,7 @@ pub(crate) struct Probe { pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, - pub(crate) nat_mode: NatMode, + pub(crate) gate: SourceGate, } /// The VPC discriminant a `u8` selector draws: one of the generated VPCs, or the bogus one. @@ -741,7 +728,11 @@ impl ProbeSpec { Probe { src_vpcd: vpcd_from_sel(self.vni_sel), dst_vpcd: self.revalidate_dst.map(vpcd_from_sel), - nat_mode: self.revalidate_nat.map(NatSel::requirement), + gate: if self.revalidate_port_fw { + SourceGate::PortFwdReply + } else { + SourceGate::Ungated + }, src_ip: block_addr( self.src_block % nblocks, self.src_host, diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index aa011d543b..1db2e1fb4d 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -29,7 +29,7 @@ pub use context::{ FlowFilterContext, FlowFilterContextReader, FlowFilterContextReaderFactory, FlowFilterContextWriter, }; -use context::{LookupInput, LookupResult}; +use context::{LookupInput, LookupResult, SourceGate}; pub struct FlowFilter { name: String, @@ -118,7 +118,7 @@ impl FlowFilter { genid: i64, ) -> Classification { let nfi = &self.name; - let (mut revalidation_dst_vpcd, mut revalidation_nat_mode) = (None, None); + let (mut revalidation_dst_vpcd, mut revalidation_gate) = (None, SourceGate::Ungated); let attached_flow = FlowSummary::from_meta(packet.meta()); if let Some(flow_summary) = attached_flow.as_ref() { // Bypass flow-filter if packet has up-to-date active flow-info @@ -126,7 +126,7 @@ impl FlowFilter { Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow_summary); return Classification::Bypassed; } - (revalidation_dst_vpcd, revalidation_nat_mode) = + (revalidation_dst_vpcd, revalidation_gate) = self.flow_revalidation_data(flow_summary, genid); } @@ -152,7 +152,7 @@ impl FlowFilter { .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }), - nat_mode: revalidation_nat_mode, + gate: revalidation_gate, }; Classification::Lookup { input, @@ -297,24 +297,26 @@ impl FlowFilter { &self, flow_summary: &FlowSummary, genid: i64, - ) -> (Option, Option) { + ) -> (Option, SourceGate) { // The only case when we need re-validation is when we have an active flow with an outdated // genid. If this is not the case, return None. if flow_summary.flow_info.status() != FlowStatus::Active || flow_summary.flow_info.genid() >= genid { - return (None, None); + return (None, SourceGate::Ungated); } if flow_summary.needs_masquerade { // If we need revalidation and the flow is masqueraded, we may need the destination VPC // id from the flow to look up for reverse traffic's entry in the "remote" table. - (flow_summary.dst_vpcd, None) + (flow_summary.dst_vpcd, SourceGate::Ungated) } else if flow_summary.needs_port_forwarding { - // In we need revalidation and the flow is with port-forwarding, we may need the NAT - // mode from the flow to look up for reverse traffic's entry in the "local" table. - (None, Some(NatRequirement::PortForwarding)) + // We need revalidation and the flow is with port-forwarding, although we don't know if + // it's on the source (reply traffic) or destination side (forward traffic). In doubt, + // turn on the gate to enable looking up for entries for reply port-forwarding traffic + // in the "local"-side context table. + (None, SourceGate::PortFwdReply) } else { - (None, None) + (None, SourceGate::Ungated) } } @@ -409,21 +411,6 @@ impl NatRequirement { VpcExposeNatConfig::PortForwarding(_) => Some(Self::PortForwarding), } } - - fn as_u8(&self) -> u8 { - match self { - NatRequirement::Static => 1, - NatRequirement::Masquerade => 2, - NatRequirement::PortForwarding => 3, - } - } - - fn convert_option(opt: Option) -> u8 { - match opt { - Some(r) => r.as_u8(), - None => 0, - } - } } pub(crate) type NatMode = Option; diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index e9a360bd61..433286b67a 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -5,6 +5,7 @@ #![cfg(test)] +use crate::context::SourceGate; use crate::context::{FlowFilterContext, FlowFilterContextWriter}; use crate::fuzz_gen::Probe; use crate::test_utils::{ @@ -1390,7 +1391,7 @@ fn probe_from_packet(pkt: &Packet, src_vpcd: VpcDiscriminant) -> Opt src_vpcd, // These packets belong to no flow, so they don't need flow revalidation info. dst_vpcd: None, - nat_mode: None, + gate: SourceGate::Ungated, src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -1428,7 +1429,7 @@ fn probe_packet(probe: &Probe) -> Option<(Packet, Probe)> { // carries never reaches the lookup: the oracle must not try to lookup for revalidation info // that the NF cannot see, so we clear revalidation info. probe.dst_vpcd = None; - probe.nat_mode = None; + probe.gate = SourceGate::Ungated; if let Some((sport, dport)) = probe.ports.as_mut() { *sport = (*sport).max(1); *dport = (*dport).max(1); From d286875b02c6103f0831442a1039d1d6c589d460 Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Fri, 7 Aug 2026 19:20:48 +0200 Subject: [PATCH 7/8] feat(flow-filter): misc nits * get remote vni from peering, avoding unnecessary lookup. * avoid unnecessary if let else Signed-off-by: Fredi Raspall --- flow-filter/src/context/tables.rs | 2 +- flow-filter/src/lib.rs | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index cba0536f90..b6a2b4113e 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -573,7 +573,7 @@ impl RuleSet { for vpc in overlay.vpc_table().values() { let src_vni = vpc.vni(); for peering in vpc.peerings() { - let remote_vni = overlay.vpc_table().get_remote_vni(peering); + let remote_vni = peering.remote_vni(); let remote_vpcd = VpcDiscriminant::from_vni(remote_vni); let default_ip = || { if peering.is_v4() { diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index 1db2e1fb4d..ee9acc5141 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -382,9 +382,7 @@ struct FlowSummary { impl FlowSummary { fn from_meta(meta: &PacketMeta) -> Option { - let Some(flow_info) = &meta.flow_info else { - return None; - }; + let flow_info = meta.flow_info.as_ref()?; let locked_info = flow_info.locked.read(); Some(Self { genid: flow_info.genid(), From d480f4fe6c1453b82c528adba068bd8e7d7f8f0f Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Fri, 7 Aug 2026 19:31:33 +0200 Subject: [PATCH 8/8] feat(flow-filter): sanitize FlowSummary Make sure FlowSummary always summarizes a correct flow by requiring the destination VPC to be always known. With the prior code, a summary could be wrong and then be subjected to re-validation. With this patch, such a flow would be ignored. Signed-off-by: Fredi Raspall --- flow-filter/src/lib.rs | 23 ++++++++++------------- flow-filter/src/tests.rs | 23 ++++++++++++----------- 2 files changed, 22 insertions(+), 24 deletions(-) diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index ee9acc5141..1674eb8309 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -269,7 +269,7 @@ impl FlowFilter { return false; } let (nfi, flowkey) = (&self.name, flow_summary.flow_info.flowkey()); - if flow_summary.dst_vpcd != Some(new_dst_vpcd) { + if flow_summary.dst_vpcd != new_dst_vpcd { debug!("{nfi}: Outdated flow {flowkey} (new dst: {new_dst_vpcd}) will be invalidated."); return true; } @@ -308,7 +308,7 @@ impl FlowFilter { if flow_summary.needs_masquerade { // If we need revalidation and the flow is masqueraded, we may need the destination VPC // id from the flow to look up for reverse traffic's entry in the "remote" table. - (flow_summary.dst_vpcd, SourceGate::Ungated) + (Some(flow_summary.dst_vpcd), SourceGate::Ungated) } else if flow_summary.needs_port_forwarding { // We need revalidation and the flow is with port-forwarding, although we don't know if // it's on the source (reply traffic) or destination side (forward traffic). In doubt, @@ -338,18 +338,10 @@ impl FlowFilter { return None; } - let Some(dst_vpcd) = flow_summary.dst_vpcd else { - debug!( - "{nfi}: Flow information does not specify destination VPC. This is a bug. Ignoring it..." - ); - flow_summary.flow_info.invalidate_pair(); - return None; - }; - // 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) + Some(flow_summary.dst_vpcd) } } @@ -374,7 +366,7 @@ impl NetworkFunction for FlowFilter { #[derive(Debug, Clone)] struct FlowSummary { genid: i64, - dst_vpcd: Option, + dst_vpcd: VpcDiscriminant, needs_masquerade: bool, needs_port_forwarding: bool, flow_info: Arc, @@ -384,9 +376,14 @@ impl FlowSummary { fn from_meta(meta: &PacketMeta) -> Option { let flow_info = meta.flow_info.as_ref()?; let locked_info = flow_info.locked.read(); + let Some(dst_vpcd) = locked_info.dst_vpcd else { + debug!("Flow info lacks destination VPC. This is a bug. Invalidating flow.."); + flow_info.invalidate_pair(); + return None; + }; Some(Self { genid: flow_info.genid(), - dst_vpcd: locked_info.dst_vpcd, + dst_vpcd, needs_masquerade: locked_info.nat_state.is_some(), needs_port_forwarding: locked_info.port_fw_state.is_some(), flow_info: flow_info.clone(), diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index 433286b67a..091f7c1627 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -1121,23 +1121,24 @@ struct InvalidationCase { meta_port_forwarding: bool, has_flow: bool, genid: GenidRel, - /// `Some(true)`: the flow's destination equals the route's; `Some(false)`: a different one; - /// `None`: the flow records no destination. - flow_dst_matches: Option, + /// `true`: the flow's destination equals the route's; `false`: a different one. A flow that + /// records no destination at all cannot reach this decision: `FlowSummary::from_meta` + /// invalidates it and reports no summary, which the `has_flow: false` cases already cover. + flow_dst_matches: bool, flow_masquerade: bool, flow_port_forwarding: bool, } // The specification: a flow is invalidated iff it comes from a DIFFERENT config generation // (older or newer -- only an equal genid is trusted) AND the filter can prove it stale: the -// destination changed (or was never recorded), a stateful-NAT requirement appeared or -// disappeared, or the route no longer needs state at all. Anything else is deferred to the -// stateful NFs, which own the state's validity. +// destination changed, a stateful-NAT requirement appeared or disappeared, or the route no longer +// needs state at all. Anything else is deferred to the stateful NFs, which own the state's +// validity. fn expected_invalidation(case: &InvalidationCase) -> bool { if !case.has_flow || matches!(case.genid, GenidRel::Same) { return false; } - case.flow_dst_matches != Some(true) + !case.flow_dst_matches || case.meta_masquerade != case.flow_masquerade || case.meta_port_forwarding != case.flow_port_forwarding || (!case.meta_masquerade && !case.meta_port_forwarding) @@ -1174,10 +1175,10 @@ fn invalidation_decision_matches_spec() { GenidRel::Same => GENID, GenidRel::Newer => GENID + 3, }, - dst_vpcd: match case.flow_dst_matches { - Some(true) => Some(route_dst), - Some(false) => Some(vpcd(300)), - None => None, + dst_vpcd: if case.flow_dst_matches { + route_dst + } else { + vpcd(300) }, needs_masquerade: case.flow_masquerade, needs_port_forwarding: case.flow_port_forwarding,