From 6ffadfe39f510a5edae76e4e56e20668f5770620 Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Tue, 4 Aug 2026 12:52:58 +0200 Subject: [PATCH 1/6] feat(flow-filter): let masqueraded traffic through if flow If a packet hits a valid flow whose generation id does not match the current, and the flow is masquerading, let the packet use that flow even if outdated. Otherwise, the flow filter would drop packets belonging to still-allowed flows unnecessarily. The flow filter does not have the knowledge that those flows should be allowed or not. It is the masquerade stage that is responsible for that. Also, rename variable and change debug! for error! in a case that should never occur. Signed-off-by: Fredi Raspall --- flow-filter/src/lib.rs | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index c2250bde67..760add04d6 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -13,7 +13,7 @@ use net::packet::{DoneReason, Packet, PacketMeta, VpcDiscriminant}; use pipeline::{NetworkFunction, PipelineData}; use std::num::NonZero; use tracectl::trace_target; -use tracing::debug; +use tracing::{debug, error}; trace_target!("flow-filter", LevelFilter::INFO, &["pipeline"]); @@ -118,11 +118,11 @@ impl FlowFilter { genid: i64, ) -> Classification { let nfi = &self.name; - let attached_flow = FlowSummary::from_meta(packet.meta()); - if let Some(flow_summary) = attached_flow.as_ref() { + let flow_summary = FlowSummary::from_meta(packet.meta()); + if let Some(summary) = flow_summary.as_ref() { // Bypass flow-filter if packet has up-to-date active flow-info - if let Some(dst_vpcd) = self.dst_vpcd_from_valid_flow(flow_summary, genid) { - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow_summary); + if let Some(dst_vpcd) = self.dst_vpcd_from_valid_flow(summary, genid) { + Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, summary); return Classification::Bypassed; } } @@ -151,7 +151,7 @@ impl FlowFilter { }; Classification::Lookup { input, - flow_summary: attached_flow, + flow_summary, } } @@ -335,7 +335,10 @@ impl FlowFilter { return None; } let flow_genid = flow_summary.flow_info.genid(); - if flow_genid < genid { + if flow_genid < genid && !flow_summary.needs_masquerade { + // If a packet belongs to a masqueraded flow, we have to let it through, temporarily, even if the + // flow is out-dated in terms of generation Id: the flow filter does not have the knowledge to + // forbid that flow. That's the responsibility of the masquerade stage. debug!( "{nfi}: Packet has outdated flow information from a prior configuration ({flow_genid} < {genid})" ); @@ -343,15 +346,13 @@ impl FlowFilter { } let Some(dst_vpcd) = flow_summary.dst_vpcd else { - debug!( - "{nfi}: Flow information does not specify destination VPC. This is a bug. Ignoring it..." - ); + error!("{nfi}: Flow info does not specify dst VPC. This is a bug. Ignoring it..."); flow_summary.flow_info.invalidate_pair(); return None; }; - // The flow has the same generation id as the current config. Small transient period aside, - // this means that the flow is up-to-date and we can bypass the filter + // The flow has the same generation id as the current config (small transient period aside), or + // it is masquerading. So, packet can bypass the flow filter. debug!("{nfi}: Packet can bypass flow filter thanks to flow information"); Some(dst_vpcd) } From 3a1b02a42ba7aa0291d433c2983880b5a36513ba Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Tue, 4 Aug 2026 13:32:11 +0200 Subject: [PATCH 2/6] feat(flow-filter): adjust tests Signed-off-by: Fredi Raspall --- flow-filter/src/tests.rs | 37 +++++++++++++++++++++++++++++++++---- 1 file changed, 33 insertions(+), 4 deletions(-) diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index 2e2c84107b..376cc5b524 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -13,6 +13,7 @@ use crate::test_utils::{ vpcd, }; use concurrency::sync::Arc; +use config::GenId; use lpm::prefix::L4Protocol; use net::FlowKey; use net::buffer::TestBuffer; @@ -625,11 +626,11 @@ 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. + // The packet hits a flow that is masquerading. The flow-filter will not set the verdict but be bypassed. 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)); - assert_eq!(flow.status(), FlowStatus::Cancelled); + assert_eq!(out.get_done(), None); + assert_eq!(flow.status(), FlowStatus::Active); } #[test] @@ -664,7 +665,7 @@ fn port_forwarding_reply_without_flow_is_filtered() { } #[test] -fn stateful_flow_does_not_survive_peering_removal() { +fn masquerade_flow_is_left_untouched_on_config_removal() { // The peering is gone from the new config: even an active, state-consistent flow must not let // reply traffic through (stage 1 finds no marker to trust), and the flow pair is invalidated. let (mut flow_filter, writer) = make_flow_filter(source_nat_context()); @@ -676,6 +677,34 @@ fn stateful_flow_does_not_survive_peering_removal() { ); let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), None); + assert_eq!(flow.status(), FlowStatus::Active); +} + +#[test] +fn portfw_flow_does_not_survive_peering_removal() { + const GENID: GenId = 5; + let (mut flow_filter, writer) = make_flow_filter(dst_port_forwarding_context()); + set_genid(&mut flow_filter, GENID); + + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("10.0.0.1"), v4("80.0.0.5"), 5678, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + flow.set_genid(GENID); + let second_packet = p.clone(); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), None); + assert_eq!(flow.status(), FlowStatus::Active); + + // config is updated, new packet (second_packet) arrives, still pointing to non-updated flow + // the packet is dropped since its flow is out-dated and it does not bypass the flow filter, + // which decides to drop it. + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, GENID + 1); + + let out = run(&mut flow_filter, second_packet); assert_eq!(out.get_done(), Some(DoneReason::Filtered)); assert_eq!(flow.status(), FlowStatus::Cancelled); } From d87aa1d2e7c06668fab00d8d69b3543ac06986e7 Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Tue, 4 Aug 2026 14:51:01 +0200 Subject: [PATCH 3/6] feat(flow-filter): use proper type for Route Avoids ordering issues in NatMode and duplicated test type. Signed-off-by: Fredi Raspall --- flow-filter/src/context/fuzz.rs | 3 +- flow-filter/src/context/tables.rs | 40 ++++++++++++---- flow-filter/src/context/tests.rs | 79 +++++++++++++------------------ flow-filter/src/lib.rs | 6 ++- 4 files changed, 73 insertions(+), 55 deletions(-) diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index 8d790713ae..a2ac2ca784 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -13,6 +13,7 @@ use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult}; use crate::NatRequirement; +use crate::context::tables::Route; use crate::fuzz_gen::{OverlaySpec, Probe, ProbeSpec, bogus_vpcd}; use concurrency::sync::LazyLock; use concurrency::sync::atomic::{AtomicU64, Ordering}; @@ -138,7 +139,7 @@ fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { consider(&mut src_nat, (0, false), None); } match src_nat { - Some((_, src_nat)) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), + Some((_, src_nat)) => LookupResult::Route(Route::new(dst_vpcd, dst_nat, src_nat)), None => LookupResult::SourceMiss(dst_vpcd), } } diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index d8e98fabf5..96291f89a1 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -56,7 +56,27 @@ use tracing::debug; /// A resolved route: destination VPC, destination NAT mode, source NAT mode. All `Copy`, so batch /// results can be extracted and the context guard dropped before packet metadata is mutated. -type Route = (VpcDiscriminant, NatMode, NatMode); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Route { + pub(crate) dst_vpcd: VpcDiscriminant, + pub(crate) dst_nat_mode: NatMode, + pub(crate) src_nat_mode: NatMode, +} +impl Route { + #[must_use] + pub(crate) fn new( + dst_vpcd: VpcDiscriminant, + dst_nat_mode: NatMode, + src_nat_mode: NatMode, + ) -> Self { + Self { + dst_vpcd, + dst_nat_mode, + src_nat_mode, + } + } +} /// One lookup outcome. The two miss variants are distinct because the NF's fallback differs: /// a destination miss means no peering covers the packet at all (drop, fail closed), while a @@ -667,9 +687,11 @@ impl FlowFilterContext { src_ip, src_port, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) - } + Some(nat_mode) => LookupResult::Route(Route::new( + verdict.dst_vpcd, + verdict.nat_mode, + *nat_mode, + )), None => LookupResult::SourceMiss(verdict.dst_vpcd), } } @@ -689,9 +711,11 @@ impl FlowFilterContext { src_ip, src_port, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) - } + Some(nat_mode) => LookupResult::Route(Route::new( + verdict.dst_vpcd, + verdict.nat_mode, + *nat_mode, + )), None => LookupResult::SourceMiss(verdict.dst_vpcd), } } @@ -806,7 +830,7 @@ fn lookup_versioned( let verdict = verdicts[pos].unwrap_or_else(|| unreachable!("hit_pos tracks Some")); out[i_chunk[pos]] = match nat_modes[hit] { Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) + LookupResult::Route(Route::new(verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) } None => LookupResult::SourceMiss(verdict.dst_vpcd), }; diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index c94dcd49bb..ae2afe5b69 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -7,21 +7,14 @@ use super::LookupResult; use super::tables::RuleRow; +use crate::context::tables::Route; use crate::test_utils::*; -use crate::{FlowFilterContext, NatMode, NatRequirement}; +use crate::{FlowFilterContext, NatRequirement}; use lpm::prefix::L4Protocol; use net::headers::Headers; use net::packet::VpcDiscriminant; use std::num::NonZero; -// Wrapper for the result of a lookup -#[derive(Debug, PartialEq, Eq)] -struct Route { - dst_vpcd: VpcDiscriminant, - dst_nat: NatMode, - src_nat: NatMode, -} - // 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( @@ -39,11 +32,7 @@ fn route( .zip(t.dst_port().map(NonZero::get)) }); match context.lookup(src_vpcd, src_ip, dst_ip, proto, ports) { - LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { - dst_vpcd, - dst_nat, - src_nat, - }), + LookupResult::Route(route) => Some(route), LookupResult::SourceMiss(_) | LookupResult::DestinationMiss => None, } } @@ -98,8 +87,8 @@ fn packet_allowed() { ) .expect("packet should be allowed"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.dst_nat, None); - assert_eq!(r.src_nat, None); + assert_eq!(r.dst_nat_mode, None); + assert_eq!(r.src_nat_mode, None); } #[test] @@ -137,8 +126,8 @@ fn default_remote_expose_is_catch_all() { ) .expect("default expose should match"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.dst_nat, None); - assert_eq!(r.src_nat, None); + assert_eq!(r.dst_nat_mode, None); + assert_eq!(r.src_nat_mode, None); } #[test] @@ -229,30 +218,30 @@ fn nat_modes_source_and_destination() { // (src NAT, dst NAT) for valid combinations. Source carries private IPs, destination carries // public IPs. let none_none = lookup("1.0.0.5", "5.0.0.10"); - assert_eq!(none_none.src_nat, None); - assert_eq!(none_none.dst_nat, None); + assert_eq!(none_none.src_nat_mode, None); + assert_eq!(none_none.dst_nat_mode, None); let static_static = lookup("2.0.0.5", "60.0.0.10"); - assert_eq!(static_static.src_nat, Some(NatRequirement::Static)); - assert_eq!(static_static.dst_nat, Some(NatRequirement::Static)); + assert_eq!(static_static.src_nat_mode, Some(NatRequirement::Static)); + assert_eq!(static_static.dst_nat_mode, Some(NatRequirement::Static)); let masq_none = lookup("3.0.0.5", "5.0.0.10"); - assert_eq!(masq_none.src_nat, Some(NatRequirement::Masquerade)); - assert_eq!(masq_none.dst_nat, None); + assert_eq!(masq_none.src_nat_mode, Some(NatRequirement::Masquerade)); + assert_eq!(masq_none.dst_nat_mode, None); let none_static = lookup("1.0.0.5", "60.0.0.10"); - assert_eq!(none_static.src_nat, None); - assert_eq!(none_static.dst_nat, Some(NatRequirement::Static)); + assert_eq!(none_static.src_nat_mode, None); + assert_eq!(none_static.dst_nat_mode, Some(NatRequirement::Static)); let static_none = lookup("2.0.0.5", "5.0.0.10"); - assert_eq!(static_none.src_nat, Some(NatRequirement::Static)); - assert_eq!(static_none.dst_nat, None); + assert_eq!(static_none.src_nat_mode, Some(NatRequirement::Static)); + assert_eq!(static_none.dst_nat_mode, None); // Masquerade source towards the default (no-NAT) destination. let masq_default = lookup("3.0.0.5", "99.0.0.10"); assert_eq!(masq_default.dst_vpcd, vpcd(200)); - assert_eq!(masq_default.src_nat, Some(NatRequirement::Masquerade)); - assert_eq!(masq_default.dst_nat, None); + assert_eq!(masq_default.src_nat_mode, Some(NatRequirement::Masquerade)); + assert_eq!(masq_default.dst_nat_mode, None); } // Destination-side NAT: a masquerade destination is filtered (cannot receive @@ -294,7 +283,7 @@ fn dst_side_nat_modes() { ) .expect("masquerade destination resolves as a marker"); assert_eq!(masq.dst_vpcd, vpcd(200)); - assert_eq!(masq.dst_nat, Some(NatRequirement::Masquerade)); + assert_eq!(masq.dst_nat_mode, Some(NatRequirement::Masquerade)); // Port-forwarding destination (matching proto + port): returned let pf = route( @@ -304,8 +293,8 @@ fn dst_side_nat_modes() { ) .expect("port forwarding destination should match"); assert_eq!(pf.dst_vpcd, vpcd(200)); - assert_eq!(pf.dst_nat, Some(NatRequirement::PortForwarding)); - assert_eq!(pf.src_nat, None); + assert_eq!(pf.dst_nat_mode, Some(NatRequirement::PortForwarding)); + assert_eq!(pf.src_nat_mode, None); // Port-forwarding destination, wrong port: no match. assert_eq!( @@ -334,7 +323,7 @@ fn protocol_awareness() { ] { let r = route(&ctx, vpcd(100), &headers).expect("plain expose matches any protocol"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.dst_nat, None); + assert_eq!(r.dst_nat_mode, None); } // TCP packet matches the TCP-only port-forwarding destination @@ -345,8 +334,8 @@ fn protocol_awareness() { ) .expect("TCP port-forwarding destination should match"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.src_nat, None); - assert_eq!(r.dst_nat, Some(NatRequirement::PortForwarding)); + assert_eq!(r.src_nat_mode, None); + assert_eq!(r.dst_nat_mode, Some(NatRequirement::PortForwarding)); // TCP-only port forwarding: a UDP packet in the same range does not match assert_eq!( @@ -391,8 +380,8 @@ fn source_default_expose_is_catch_all() { ) .expect("local default expose should match the source"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.src_nat, None); - assert_eq!(r.dst_nat, None); + assert_eq!(r.src_nat_mode, None); + assert_eq!(r.dst_nat_mode, None); } // ------------------------------------------------------------------------------------------------- @@ -423,7 +412,7 @@ fn port_forwarding_any_protocol_matches_tcp_and_udp() { ] { let r = route(&ctx, vpcd(100), &headers).expect("any-protocol port forwarding matches"); assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.dst_nat, Some(NatRequirement::PortForwarding)); + assert_eq!(r.dst_nat_mode, Some(NatRequirement::PortForwarding)); } } @@ -462,8 +451,8 @@ fn source_port_forwarding_is_excluded_and_falls_back_to_masquerade() { ) .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); + assert_eq!(r.src_nat_mode, Some(NatRequirement::Masquerade)); + assert_eq!(r.dst_nat_mode, None); } // ------------------------------------------------------------------------------------------------- @@ -552,8 +541,8 @@ fn discrepancy_overlapping_contiguous_prefixes() { ) .expect("request: single matching destination in table should be found based on src/dst IPs"); assert_eq!(r.dst_vpcd, vpcd(300)); - assert_eq!(r.src_nat, None); - assert_eq!(r.dst_nat, None); + assert_eq!(r.src_nat_mode, None); + assert_eq!(r.dst_nat_mode, None); let r = route( &ctx, @@ -562,8 +551,8 @@ fn discrepancy_overlapping_contiguous_prefixes() { ) .expect("reply: single matching destination in table should be found based on src/dst IPs"); assert_eq!(r.dst_vpcd, vpcd(100)); - assert_eq!(r.src_nat, None); - assert_eq!(r.dst_nat, None); + assert_eq!(r.src_nat_mode, None); + assert_eq!(r.dst_nat_mode, None); // Check there are no /32 prefixes in the remote-side rules for RuleRow { rule, .. } in ctx.remote_v4.rules() { diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index 760add04d6..1fdaa852fa 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -172,7 +172,7 @@ impl FlowFilter { genid: i64, ) { let nfi = &self.name; - let (dst_vpcd, dst_nat_mode, src_nat_mode) = match result { + let route = match result { LookupResult::Route(route) => route, LookupResult::SourceMiss(dst_vpcd) => { // Port-forwarding sources are deliberately absent from the local tables; reply @@ -197,6 +197,10 @@ impl FlowFilter { } }; + let dst_nat_mode = route.dst_nat_mode; + let src_nat_mode = route.src_nat_mode; + let dst_vpcd = route.dst_vpcd; + // 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. From 4fdaa9ee5c2eeaee64ba82be07046781b9eda00e Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Mon, 27 Jul 2026 19:12:25 +0200 Subject: [PATCH 4/6] feat(net): add FlowInfoFlag for master flow in pair Flows live in pairs in the flow table. When exporting them, we will want to export them "in pairs", because having only one of them is of little use to forward the associated traffic. E.g. if flows F1 and F2 are related, we may export some (F1,F2). However, there is no notion of "forward" or "reverse". Both F1 and F2 are currently equally important. This creates a problem to export them in pairs: we may create some export object for pair (F1, F2) when scanning the flow table, but then need a way to know that (F2, F1) needs not be exported, to avoid sending the information twice. To solve this, we add a flag "master" to each flow and make sure that the flag is only set for one of the flows in a pair. This way, we only need to export "master" flows. The "master" flag has no semantic other than that, but could always be set to the flow corresponding to the packet that triggered the creation of the flow pair. Signed-off-by: Fredi Raspall --- net/src/flows/flow_info.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/net/src/flows/flow_info.rs b/net/src/flows/flow_info.rs index 23e86981df..1a4e04ecf0 100644 --- a/net/src/flows/flow_info.rs +++ b/net/src/flows/flow_info.rs @@ -140,12 +140,18 @@ impl From for AtomicFlowStatus { bitflags! { #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] pub struct FlowInfoFlags: u8 { - const REQ_STATIC_NAT_SRC = 0b0000_0001; /* Packet requires static NAT (source) */ - const REQ_STATIC_NAT_DST = 0b0000_0010; /* Packet requires static NAT (destination) */ + const IS_PAIR_MASTER = 0b0000_0001; /* the flow is the master of a pair */ + const REQ_STATIC_NAT_SRC = 0b0000_0010; /* Packet requires static NAT (source) */ + const REQ_STATIC_NAT_DST = 0b0000_0100; /* Packet requires static NAT (destination) */ } } impl FlowInfoFlags { + #[must_use] + pub const fn is_pair_master(&self) -> bool { + self.contains(FlowInfoFlags::IS_PAIR_MASTER) + } + #[must_use] pub const fn requires_static_nat_src(self) -> bool { self.contains(FlowInfoFlags::REQ_STATIC_NAT_SRC) @@ -282,9 +288,9 @@ impl FlowInfo { pub fn related_pair( expires_at: Instant, key1: FlowKey, - flags1: FlowInfoFlags, + mut flags1: FlowInfoFlags, key2: FlowKey, - flags2: FlowInfoFlags, + mut flags2: FlowInfoFlags, ) -> (Arc, Arc) { // keys MUST differ debug_assert!( @@ -292,6 +298,10 @@ impl FlowInfo { "Attempted to build two flows with identical key {key1}" ); + // make sure only one of the pairs is master + flags1.insert(FlowInfoFlags::IS_PAIR_MASTER); + flags2.remove(FlowInfoFlags::IS_PAIR_MASTER); + let mut one: Arc> = Arc::new_uninit(); let mut two: Arc> = Arc::new_uninit(); From a3b3183f94972355028172459ee69d7051223c36 Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Tue, 4 Aug 2026 20:27:22 +0200 Subject: [PATCH 5/6] refactor(flow-filter): reshape the flow filter Summary of changes (conceptual): * let stage-1 not have entries where remote requires masquerade as that would provide multiple answers due to overlap. * let stage-2 have all (local) entries, including port-forwarding * revisit flow-filter logic: - packets with no flow or non-active flow are always checked. - only packets with active, up-to-date flows can bypass filter. - packets with active but non-up-to-date flows cannot bypass since we don't know if the flow would still be valid with a new configuration. So, the packet/flow needs to be re-evaluated. As before, the re-evaluation may not be complete because the flow filter does not have full visibility to the NF state. But it should disqualify when it is possible to ensure that packets are delivered to the intended recipient and marked to get the treatment according to the configuration. - A packet may not always allow us to validate a flow, because it may come from the direction where the flow-filter lacks the knowledge to unambiguosly tell what to do. In those cases, when attempting to validate an existing flow created under a prior config, we ask the flow filter to tell us if the packet that would have initiated that flow would still be allowed; instead of checking the packet that we received. If that prior packet would be allowed and the flow that the received packet rides on now is compatible with that verdict, the flow is considered valid and guided to the respective NFs, which have the last word to tell if the flow is valid or not. There is no duplication here: the flow-filter has to determine what NFs need to process the packet. Code changes: - removed the zipping of the two iterators and use a single index. - use the flowsummary to annotate if the flow is "master" (initiator) and let it include src vpcd. - add logic to build input lookup key from master flow when needed. - propagate the input key to the result processing logic. - rework the logic to accept a flow/packet depending on the case. - Adapt fuzzer, tests and some table utils (Claude). Signed-off-by: Fredi Raspall --- Cargo.lock | 1 + flow-filter/Cargo.toml | 1 + flow-filter/src/context/display.rs | 21 +- flow-filter/src/context/fuzz.rs | 41 ++- flow-filter/src/context/mod.rs | 2 +- flow-filter/src/context/tables.rs | 118 ++++++-- flow-filter/src/context/tests.rs | 87 +++++- flow-filter/src/fuzz_gen.rs | 36 ++- flow-filter/src/lib.rs | 397 +++++++++++++++++---------- flow-filter/src/tests.rs | 427 ++++++++++++++++++++++------- 10 files changed, 810 insertions(+), 321 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 630c9f6dd8..0c599383ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1453,6 +1453,7 @@ dependencies = [ "indenter", "linkme", "tracing", + "tracing-test", ] [[package]] diff --git a/flow-filter/Cargo.toml b/flow-filter/Cargo.toml index 00c9063fec..4822fb5c5f 100644 --- a/flow-filter/Cargo.toml +++ b/flow-filter/Cargo.toml @@ -33,3 +33,4 @@ bolero = { workspace = true, features = ["std"] } dpdk = { workspace = true, features = ["test"] } lpm = { workspace = true, features = ["testing"] } net = { workspace = true, features = ["builder"] } +tracing-test = { workspace = true } diff --git a/flow-filter/src/context/display.rs b/flow-filter/src/context/display.rs index b9561c9bc6..0085a76d49 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, Route}; impl crate::NatRequirement { fn label(self) -> &'static str { @@ -24,6 +24,25 @@ impl std::fmt::Display for crate::NatRequirement { } } +fn nat_mode_label(mode: crate::NatMode) -> &'static str { + match mode { + Some(nat) => nat.label(), + None => "--", + } +} + +impl std::fmt::Display for Route { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "dst-vpcd: {} local: {} remote: {}", + self.dst_vpcd, + nat_mode_label(self.src_nat_mode), + nat_mode_label(self.dst_nat_mode), + ) + } +} + // ------------------------------------------------------------------------------------------------- // 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 a2ac2ca784..a842668468 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -21,6 +21,7 @@ use config::external::overlay::ValidatedOverlay; use lpm::prefix::{IpPrefix, L4Protocol, Prefix, PrefixWithOptionalPorts}; use net::ip::NextHeader; use net::packet::VpcDiscriminant; +use std::fmt; use std::net::IpAddr; // ------------------------------------------------------------------------------------------------- @@ -51,13 +52,19 @@ fn prefix_allows(prefix: &PrefixWithOptionalPorts, ip: IpAddr, port: u16) -> boo .is_none_or(|r| r.start() <= port && port <= r.end()) } -/// Rule precedence, in structural form: longest prefix first, port forwarding breaking +/// Stage-1 precedence, in structural form: longest prefix first, port forwarding breaking /// equal-length ties. Mirrors `rule_priority` without sharing its encoding. type Precedence = (u8, bool); +/// Stage-2 precedence: non-port-forwarding band first (`true` sorts above `false`), then longest +/// prefix within a band. Mirrors `local_rule_priority` -- note the fields are ordered the opposite +/// way round from [`Precedence`], because there the band dominates prefix length rather than +/// breaking ties in it. +type LocalPrecedence = (bool, u8); + /// Keep the strictly-better candidate; equal precedence between candidates that can match the /// same packet is a generator invariant violation, so fail loudly rather than pick one. -fn consider(best: &mut Option<(Precedence, T)>, precedence: Precedence, value: T) { +fn consider(best: &mut Option<(P, T)>, precedence: P, value: T) { match best { Some((current, _)) if *current == precedence => { panic!("ambiguous match at precedence {precedence:?}: generator invariant violated") @@ -82,11 +89,18 @@ fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { 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. + // excluded: they cannot receive connections, so they are withheld from the remote tables + // entirely and such a destination is a plain miss. A default expose acts as a /0 of the + // peering's IP version. let mut verdict: Option<(Precedence, (VpcDiscriminant, Option))> = None; for peering in src_vpc.peerings() { let dst_vpcd = VpcDiscriminant::from_vni(peering.remote_vni()); - for expose in peering.remote().valexp() { + for expose in peering + .remote() + .valexp() + .iter() + .filter(|expose| !expose.has_masquerade()) + { if !proto_allows(expose.nat_proto(), probe.proto) { continue; } @@ -108,20 +122,17 @@ fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { return LookupResult::DestinationMiss; }; - // 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. + // Stage 2: the source against that peering's private prefixes. Every expose participates, + // port-forwarding included: the tables carry them so that the NF sees a source NAT mode of + // `PortForwarding` (which it gates on flow state) instead of an indistinguishable source + // miss. 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")); - let mut src_nat: Option<(Precedence, Option)> = None; - for expose in peering - .local() - .valexp() - .iter() - .filter(|expose| expose.can_init_connection()) - { + let mut src_nat: Option<(LocalPrecedence, Option)> = None; + for expose in peering.local().valexp().iter() { if !proto_allows(expose.nat_proto(), probe.proto) { continue; } @@ -129,14 +140,14 @@ fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { if prefix_allows(prefix, probe.src_ip, sport) { consider( &mut src_nat, - (prefix.prefix().length(), false), + (!expose.has_port_forwarding(), prefix.prefix().length()), NatRequirement::from_expose(expose), ); } } } if peering.local().has_default_expose() && probe.src_ip.is_ipv4() == peering.is_v4() { - consider(&mut src_nat, (0, false), None); + consider(&mut src_nat, (true, 0), None); } match src_nat { Some((_, src_nat)) => LookupResult::Route(Route::new(dst_vpcd, dst_nat, src_nat)), diff --git a/flow-filter/src/context/mod.rs b/flow-filter/src/context/mod.rs index 88cc7c8300..7f2892d01b 100644 --- a/flow-filter/src/context/mod.rs +++ b/flow-filter/src/context/mod.rs @@ -15,8 +15,8 @@ mod tables; #[cfg(test)] mod tests; -pub use tables::FlowFilterContext; use tables::PRODUCTION_BACKEND; +pub use tables::{FlowFilterContext, Route}; pub(crate) use tables::{LookupInput, LookupResult}; impl TryFrom<&ValidatedOverlay> for FlowFilterContext { diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index 96291f89a1..b5557e3086 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -18,13 +18,19 @@ //! 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. +//! Neither direction of a stateful-NAT session can be settled by prefix matching +//! alone, and the two directions are handled differently: +//! +//! - Masquerade destinations cannot accept new connections, so they are withheld +//! from the remote tables entirely: such a destination is an ordinary +//! [`LookupResult::DestinationMiss`], and the NF resolves reply traffic on an +//! established masquerade flow from there. A [`Verdict`] can therefore never +//! carry [`NatRequirement::Masquerade`]. +//! - Port-forwarding sources cannot initiate connections, but they *are* kept in +//! the local tables, so that reply traffic from one is distinguishable from a +//! source no expose covers. They rank below every other source rule (see +//! [`local_rule_priority`]), which makes them a pure fallback, and the NF gates +//! the resulting [`NatRequirement::PortForwarding`] on flow state. use crate::{NatMode, NatRequirement}; use acl::dpdk::dyn_table::predicate_to_chunks; @@ -58,7 +64,7 @@ use tracing::debug; /// results can be extracted and the context guard dropped before packet metadata is mutated. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) struct Route { +pub struct Route { pub(crate) dst_vpcd: VpcDiscriminant, pub(crate) dst_nat_mode: NatMode, pub(crate) src_nat_mode: NatMode, @@ -77,11 +83,11 @@ impl Route { } } } - -/// One lookup outcome. The two miss variants are distinct because the NF's fallback differs: -/// a destination miss means no peering covers the packet at all (drop, fail closed), while a -/// source miss can still be legitimate reply traffic from a port-forwarding-only source, whose -/// rules are deliberately absent from the local tables (the NF resolves it against flow state). +/// One lookup outcome. The two miss variants are distinct because the NF's fallback differs: a +/// destination miss can still be reply traffic on an established masquerade flow (masquerade +/// destinations are withheld from the remote tables, so the NF resolves them against flow state), +/// while a source miss means no expose covers the source at all -- port-forwarding sources are in +/// the local tables, as the lowest-priority band, so reaching here is a genuine miss (drop). #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum LookupResult { /// Both stages matched. @@ -407,6 +413,27 @@ fn rule_priority(ip_range: Prefix, port_forwarding: bool) -> u32 { ((u32::from(ip_range.length()) + 1) << 1) | u32::from(port_forwarding) } +/// Stage-2 (local) rule priority: a port-forwarding source ranks below *every* non-port-forwarding +/// source, whatever prefix lengths are involved. +/// +/// A port-forwarding expose cannot answer for connection initiation, so on the source side it is a +/// pure fallback -- consulted only when no other expose covers the source, which is exactly the +/// established-reply case the NF gates on flow state. Prefix length cannot express that on its own: +/// a forwarded host prefix is *longer* than the block it is nested in, so plain +/// longest-prefix-match would let it capture traffic that a covering masquerade or plain expose +/// must answer for. Hence a band bit above everything [`rule_priority`] can produce (its maximum is +/// length 128 with the tie bit set, `((128 + 1) << 1) | 1` = 259), which therefore dominates it. +/// Within a band, ordering stays pure prefix-length. +fn local_rule_priority(ip_range: Prefix, port_forwarding: bool) -> u32 { + const NON_PORT_FORWARDING_BAND: u32 = 1 << 9; + rule_priority(ip_range, false) + | if port_forwarding { + 0 + } else { + NON_PORT_FORWARDING_BAND + } +} + /// Lower a stage-1 (remote) rule into the v4 or v6 bucket according to its prefix. #[allow(clippy::too_many_arguments)] // internal builder; grouping the fields would not aid clarity fn emit_remote( @@ -466,9 +493,7 @@ fn emit_local( proto: MaskSpec, 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. - let priority = rule_priority(ip_range, false); + let priority = local_rule_priority(ip_range, action == Some(NatRequirement::PortForwarding)); match ip_range { Prefix::IPV4(prefix) => { let rule = LocalKeyRule:: { @@ -527,12 +552,13 @@ 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. - for expose in peering.remote().valexp() { + // Stage 1: peer's public prefixes -> Verdict{dst VPC, dst NAT} + for expose in peering + .remote() + .valexp() + .iter() + .filter(|expose| !expose.has_masquerade()) + { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = Verdict { nat_mode: NatRequirement::from_expose(expose), @@ -565,14 +591,8 @@ 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()) - { + // Stage 2: source's private prefixes -> source NAT mode. + for expose in peering.local().valexp().iter() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = NatRequirement::from_expose(expose); for prefix in expose.ips() { @@ -918,4 +938,42 @@ mod unit_tests { assert!(prio_a >= 1, "priority must be a valid rte_acl priority"); }); } + + /// `local_rule_priority` puts port-forwarding sources in a strictly lower band: a + /// non-port-forwarding rule outranks a port-forwarding one at *every* pair of prefix lengths + /// (which plain longest-prefix-match would not do -- a forwarded host prefix is longer than the + /// block it nests in), and within a band longest prefix still wins. Every produced value must + /// still install as an rte_acl priority. + #[test] + fn local_priority_ranks_port_forwarding_below_every_other_source() { + use lpm::prefix::{IpPrefix, Ipv6Prefix}; + use std::net::Ipv6Addr; + let prefix_of_len = |len: u8| { + Prefix::IPV6(Ipv6Prefix::new(Ipv6Addr::UNSPECIFIED, len).expect("valid length")) + }; + bolero::check!() + .with_type::<(u8, bool, u8, bool)>() + .for_each(|&(len_a, fw_a, len_b, fw_b)| { + let (len_a, len_b) = (len_a % 129, len_b % 129); + let prio_a = local_rule_priority(prefix_of_len(len_a), fw_a); + let prio_b = local_rule_priority(prefix_of_len(len_b), fw_b); + assert_eq!( + prio_a.cmp(&prio_b), + (!fw_a, len_a).cmp(&(!fw_b, len_b)), + "local priority order diverges from (non-port-forwarding, length) order for \ + ({len_a}, {fw_a}) vs ({len_b}, {fw_b})", + ); + if fw_a && !fw_b { + assert!( + prio_a < prio_b, + "a /{len_a} port-forwarding source must lose to a /{len_b} non-forwarding \ + source, whatever the lengths", + ); + } + assert!( + i32::try_from(prio_a).is_ok_and(|p| Priority::new(p).is_ok()), + "local priority {prio_a} is not installable as an rte_acl priority", + ); + }); + } } diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index ae2afe5b69..f763c96120 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -15,13 +15,13 @@ use net::headers::Headers; use net::packet::VpcDiscriminant; use std::num::NonZero; -// Extract the 5-tuple from headers (as the pipeline does) and run the route lookup for a packet +// Extract the 5-tuple from headers (as the pipeline does) and run the lookup for a packet // originating from a given source VPC. -fn route( +fn lookup_result( context: &FlowFilterContext, src_vpcd: VpcDiscriminant, headers: &Headers, -) -> Option { +) -> LookupResult { let net = headers.net().unwrap(); let src_ip = net.src_addr(); let dst_ip = net.dst_addr(); @@ -31,7 +31,17 @@ fn route( .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }); - match context.lookup(src_vpcd, src_ip, dst_ip, proto, ports) { + context.lookup(src_vpcd, src_ip, dst_ip, proto, ports) +} + +// The resolved route, collapsing both misses into `None`. Tests that need to tell the two apart +// (the NF's fallback differs) use `lookup_result` instead. +fn route( + context: &FlowFilterContext, + src_vpcd: VpcDiscriminant, + headers: &Headers, +) -> Option { + match lookup_result(context, src_vpcd, headers) { LookupResult::Route(route) => Some(route), LookupResult::SourceMiss(_) | LookupResult::DestinationMiss => None, } @@ -274,16 +284,16 @@ 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) + // Masquerade destination: no route is found let masq = route( &ctx, vpcd(100), &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_mode, Some(NatRequirement::Masquerade)); + ); + assert!(masq.is_none()); + + // (see `masquerade_destination_is_absent_from_the_remote_tables` for why, and for which + // flavour of miss the NF relies on here) // Port-forwarding destination (matching proto + port): returned let pf = route( @@ -307,6 +317,51 @@ fn dst_side_nat_modes() { ); } +// A masquerade expose cannot receive new connections, so it is deliberately absent from the +// remote (stage-1) tables. The NF depends on both halves of that, and neither is visible from +// `route_packet`, so pin them here: +// +// - No `Verdict` can carry `Masquerade`, which is why `route_packet` has no masquerade branch at +// all. If a masquerade expose is ever emitted into the remote tables again, that branch has to +// come back with it -- otherwise such a destination is silently *allowed* with no flow check. +// - The miss is a `DestinationMiss`, not a `SourceMiss`. Only the destination-miss arm resolves +// reply traffic against an established masquerade flow; a source miss drops unconditionally. +#[test] +fn masquerade_destination_is_absent_from_the_remote_tables() { + let ctx = dst_side_overlay(); + + // The overlay is v4, so the v4 remote table is where a masquerade verdict could appear. + for RuleRow { action, .. } in ctx.remote_v4.rules() { + assert_ne!( + action.nat_mode, + Some(NatRequirement::Masquerade), + "a masquerade expose reached the remote tables", + ); + } + + // The masquerade public range resolves to nothing at all -- not to a marker, and not to a + // source miss. + assert_eq!( + lookup_result( + &ctx, + vpcd(100), + &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), + ), + LookupResult::DestinationMiss, + ); + + // Only the masquerade expose is withheld: its siblings in the same peering still route, so a + // filter that dropped too much would fail here rather than pass silently. + let plain = route( + &ctx, + vpcd(100), + &build_tcp_packet(v4("10.0.0.5"), v4("90.0.0.10"), 1234, 5678), + ) + .expect("the plain expose in the same peering still routes"); + assert_eq!(plain.dst_vpcd, vpcd(200)); + assert_eq!(plain.dst_nat_mode, None); +} + // ------------------------------------------------------------------------------------------------- // L4 protocols, including ICMP. Port-forwarding restricted to TCP must not match UDP or ICMP; plain // (no-NAT) exposes match all protocols. @@ -857,16 +912,24 @@ fn display_is_identical_across_backends() { "the remote v4 section ran past its own table:\n{remote_v4}" ); + // The masquerade public range is not installed at all (see + // `masquerade_destination_is_absent_from_the_remote_tables`), so it must not be rendered: an + // operator reading this dump would otherwise believe that destination is reachable. + assert!( + !remote_v4.contains("70.0.0.0/24"), + "the masquerade public range must not appear in the remote v4 dump:\n{remote_v4}" + ); + // The index is the operator-facing precedence claim -- `[0]` is consulted first -- so the dump // must read in match order. Within one table that is longest-prefix-first: the port-forwarding - // /32 outranks the /24s it is nested among. + // /32 outranks the /24 it is nested among. let rank = |needle: &str| { remote_v4 .find(needle) .unwrap_or_else(|| panic!("{needle} missing from the remote v4 table:\n{remote_v4}")) }; assert!( - rank("80.0.0.5/32") < rank("70.0.0.0/24") && rank("80.0.0.5/32") < rank("90.0.0.0/24"), + rank("80.0.0.5/32") < rank("90.0.0.0/24"), "rules are not rendered in precedence order:\n{remote_v4}" ); } diff --git a/flow-filter/src/fuzz_gen.rs b/flow-filter/src/fuzz_gen.rs index 819ad985f3..66fc7af6a5 100644 --- a/flow-filter/src/fuzz_gen.rs +++ b/flow-filter/src/fuzz_gen.rs @@ -80,8 +80,9 @@ pub(crate) enum ExposeSpec { Masquerade, /// Masquerade plus a port-forwarding host prefix nested inside its blocks (legal overlap). MasqueradeNestingPortFw(FwProto), - /// Masquerade plus a port-forwarding block of the SAME prefix length (legal overlap; the - /// equal-length rte_acl priority tie that the port-forwarding bit must break). + /// Masquerade plus a port-forwarding block of the SAME prefix length (legal overlap). On the + /// source side this is the case that plain longest-prefix-match cannot decide, so the + /// port-forwarding priority band has to: the masquerade expose must win. MasqueradeSameLenPortFw(FwProto), PortForwarding(FwProto), /// Two port-forwarding exposes sharing prefixes and ports, distinguished only by protocol @@ -99,10 +100,11 @@ impl ExposeSpec { !matches!(self, ExposeSpec::Plain) } - /// Whether this expose gives the source side of a route an unconstrained, connection-initiating - /// match: a plain / static-nat / masquerade private block (a `/24` or `/120` with no port - /// constraint, `can_init_connection`). Port forwarding cannot initiate, so a pure - /// port-forwarding expose is not source-capable. + /// Whether this expose gives the source side of a route an unconstrained match: a plain / + /// static-nat / masquerade private block (a `/24` or `/120` with no port constraint). A pure + /// port-forwarding expose is excluded -- it is in the local tables now, but only as a host + /// prefix constrained to the forwarded ports, which a derived probe's host `.1` and ports + /// `(1, 1)` do not match. fn source_capable(self) -> bool { !matches!( self, @@ -110,18 +112,24 @@ impl ExposeSpec { ) } - /// Where a destination address this expose matches lives, or `None` if it only matches - /// port-forwarded destinations (skipped -- those need a specific public port). `Some(true)` - /// means the public block (NAT exposes translate destinations into it); `Some(false)` means the - /// private block (a plain expose's public IPs are its private IPs). + /// Where a destination address this expose matches lives, or `None` if it matches no + /// destination a derived probe can reach. `Some(true)` means the public block (NAT exposes + /// translate destinations into it); `Some(false)` means the private block (a plain expose's + /// public IPs are its private IPs). + /// + /// Two kinds yield `None`. Port-forwarded destinations need a specific public port, which a + /// derived probe does not carry. And masquerade destinations cannot receive connections at + /// all, so they are withheld from the remote tables -- including the masquerade half of the + /// composite kinds, whose port-forwarding half is only reachable at the forwarded port. fn dest_public_space(self) -> Option { match self { ExposeSpec::Plain => Some(false), - ExposeSpec::StaticNat - | ExposeSpec::Masquerade + ExposeSpec::StaticNat => Some(true), + ExposeSpec::Masquerade | ExposeSpec::MasqueradeNestingPortFw(_) - | ExposeSpec::MasqueradeSameLenPortFw(_) => Some(true), - ExposeSpec::PortForwarding(_) | ExposeSpec::PortFwProtoPair => None, + | ExposeSpec::MasqueradeSameLenPortFw(_) + | ExposeSpec::PortForwarding(_) + | ExposeSpec::PortFwProtoPair => None, } } } diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index 1fdaa852fa..586e9b1dbf 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -3,7 +3,8 @@ #![doc = include_str!("../README.md")] -use concurrency::sync::Arc; +use crate::context::Route; +use concurrency::sync::{Arc, Weak}; use config::external::overlay::vpcpeering::{ValidatedExpose, VpcExposeNatConfig}; use net::FlowKey; use net::buffer::PacketBufferMut; @@ -13,7 +14,7 @@ use net::packet::{DoneReason, Packet, PacketMeta, VpcDiscriminant}; use pipeline::{NetworkFunction, PipelineData}; use std::num::NonZero; use tracectl::trace_target; -use tracing::{debug, error}; +use tracing::{debug, error, warn}; trace_target!("flow-filter", LevelFilter::INFO, &["pipeline"]); @@ -44,6 +45,7 @@ enum Classification { /// Drop the packet (Setting the DoneReason in its metadatas is done in place). Drop, /// Needs a table lookup; carries the query and any attached flow summary (for phase C). + /// `LookupInput` may be built from the packet or the master flow in the reverse direction Lookup { input: LookupInput, flow_summary: Option, @@ -100,46 +102,67 @@ impl FlowFilter { let tables = self.tables.load(); tables.lookup_batch(&inputs, &mut results); - for (item, result) in work.iter().zip(results) { + // `work`, `inputs` and `results` are aligned by index + for (i, item) in work.iter().enumerate() { self.apply_route( &mut burst[item.idx], - result, item.flow_summary.as_ref(), - genid, + inputs[i], + results[i], ); } } - /// Phase A: decide what a single overlay packet needs. Tags bypass packets in place; returns - /// the [`LookupInput`] (plus any attached flow summary, which phase C needs) otherwise. - fn classify( + fn get_packet_flow_state( &self, - packet: &mut Packet, - genid: i64, - ) -> Classification { + packet: &Packet, + ) -> Option { let nfi = &self.name; - let flow_summary = FlowSummary::from_meta(packet.meta()); - if let Some(summary) = flow_summary.as_ref() { - // Bypass flow-filter if packet has up-to-date active flow-info - if let Some(dst_vpcd) = self.dst_vpcd_from_valid_flow(summary, genid) { - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, summary); - return Classification::Bypassed; - } + let flow_info = packet.meta().flow_info.as_ref()?; + if flow_info.status() != FlowStatus::Active { + debug!("{nfi}: Packet hit non-active flow. Will ignore flow"); + return None; } - - let Some(net) = packet.try_ip() else { - debug!("{nfi}: No IP headers found, dropping packet"); - packet.done(DoneReason::NotIp); - return Classification::Drop; - }; - let Some(src_vpcd) = packet.meta().src_vpcd else { - debug!("{nfi}: Missing source VPC discriminant, dropping packet"); - packet.done(DoneReason::Unroutable); - return Classification::Drop; + let Some(summary) = FlowSummary::from_flow_info(flow_info) else { + error!("{nfi}: Bad flow summary: missing src or dst vpc discriminant. This is a bug"); + packet.invalidate_flows(); + return None; }; + // packet hits a sane, active flow, but we have not yet checked if the flow + // is compatible with the current generation id or not + Some(summary) + } + /// Build a `LookupInput` from the flow key that led to the creation of this flow + fn build_lookup_key_from_flow(summary: &FlowSummary) -> Option { + let flow_info = summary.flow_info.related.as_ref().and_then(Weak::upgrade)?; + if !flow_info.get_flags().is_pair_master() { + error!("Related flow of a non-master flow is not the master. This is a bug"); + return None; + } + // related flow could be inactive (unlikely) + if flow_info.status() != FlowStatus::Active { + debug!("Won't use related flow: it is not active"); + return None; + } + let key = flow_info.flowkey(); let input = LookupInput { - src_vpcd, + src_vpcd: key.src_vpcd().unwrap_or_else(|| unreachable!()), + src_ip: *key.src_ip(), + dst_ip: *key.dst_ip(), + proto: key.proto(), + ports: key.ports().map(|(s, d)| (s.get(), d.get())), + }; + debug!("Will validate flow {key}"); + Some(input) + } + + /// Build a `LookupInput` from the packet. This assumes that the packet is IP and + /// that it is annotated with the src vpcd + fn build_lookup_key_from_packet(packet: &Packet) -> LookupInput { + let net = packet.try_ip().unwrap_or_else(|| unreachable!()); + LookupInput { + src_vpcd: packet.meta().src_vpcd.unwrap_or_else(|| unreachable!()), src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -148,82 +171,159 @@ impl FlowFilter { .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }), - }; - Classification::Lookup { - input, - flow_summary, } } - /// 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( + #[inline] + fn packet_is_valid(&self, packet: &mut Packet) -> bool { + let nfi = &self.name; + + // disqualify non-ip + if packet.try_ip().is_none() { + debug!("{nfi}: No IP header found, dropping packet"); + packet.done(DoneReason::NotIp); + return false; + }; + // disqualify unknown origin + if packet.meta().src_vpcd.is_none() { + debug!("{nfi}: Missing source VPC discriminant, dropping packet"); + packet.done(DoneReason::Unroutable); + return false; + }; + true + } + + /// Phase A: decide what a single overlay packet needs. Tags bypass packets in place; returns + /// the [`LookupInput`] (plus any attached flow summary, which phase C needs) otherwise. + fn classify( &self, packet: &mut Packet, - result: LookupResult, - flow_summary: Option<&FlowSummary>, genid: i64, - ) { + ) -> Classification { let nfi = &self.name; - let route = 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; + + if !self.packet_is_valid(packet) { + return Classification::Drop; + } + + // Get the flow info that the packet matched. If packet matched no flow or it did but the flow is not + // active, we ignore it and the flow filter always evaluates the packet. + // If the packet matched an active flow, there are two possiblities: + // 1) the flow is up-to-date (in terms of genid): the packet can bypass the flow filter confidently. + // 2) the flow is not up-to-date: the packet can't' bypass the flow filter since we don't know if + // the flow should still be allowed nor the current treatment. + // In case (2) it may happen that we cannot determine if the packet is "routable" nor the treatment it + // should get under a new config because it may not correspond to the flow that initiated the communication. + // In that case, we ask the flow filter if a packet (in the reverse direction) that would have initiated the + // flow would still be routable. If that packet would be denied, we know the packet/flow must be dropped. + // If such a packet would be allowed (we get a route), we trust the possibly out-dated flow if its treatment is + // compatible with the route hit by such a packet. + + // Build a flow summary from the packet: we only get one if the packet hit an active, valid flow (might be out-dated) + let flow_summary = self.get_packet_flow_state(packet); + if let Some(summary) = flow_summary.as_ref() { + let flowkey = summary.flow_info.flowkey(); + debug!("{nfi}: Packet matched active flow ({flowkey})"); + if summary.genid < genid { + let master = summary.is_master_flow; + debug!("{nfi}: Flow ({flowkey}) (master: {master}) could be out-dated"); + let input = if master { + Some(Self::build_lookup_key_from_packet(packet)) + } else { + Self::build_lookup_key_from_flow(summary) + }; + if let Some(input) = input { + Classification::Lookup { + input, + flow_summary, + } + } else { + warn!("{nfi}: Could not build lookup key from master flow. Will drop"); + packet.done(DoneReason::Unroutable); + packet.invalidate_flows(); + Classification::Drop } - debug!("{nfi}: Source not allowed towards {dst_vpcd}, dropping packet"); - packet.invalidate_flows(); - packet.done(DoneReason::Filtered); - return; + } else { + debug!("{nfi}: Flow ({flowkey}) is up-to-date. Will bypass flow-filter"); + Self::tag_for_bypass(packet.meta_mut(), summary); + Classification::Bypassed } - LookupResult::DestinationMiss => { - debug!("{nfi}: Could not determine destination VPC, dropping packet"); - packet.invalidate_flows(); - packet.done(DoneReason::Filtered); - return; + } else { + debug!("{nfi}: Packet did not match any active flow"); + let input = Self::build_lookup_key_from_packet(packet); + Classification::Lookup { + input, + flow_summary, } - }; + } + } - let dst_nat_mode = route.dst_nat_mode; - let src_nat_mode = route.src_nat_mode; - let dst_vpcd = route.dst_vpcd; - - // 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; - } + /// The NAT requirements of a route + fn route_nat_requirements(route: &Route) -> PacketMeta { + let mut meta = PacketMeta::default(); + Self::set_nat_requirements(&mut meta, route.src_nat_mode, route.dst_nat_mode); + meta + } + + fn validate_flow_from_reverse_route( + &self, + packet: &mut Packet, + flow_summary: &FlowSummary, + route: &Route, + input: LookupInput, + ) { + let nfi = &self.name; + debug!("{nfi}: Reverse flow would match route {route}"); + + // The initiating direction must point back to where the packet came from + if route.dst_vpcd != flow_summary.src_vpcd { debug!( - "{nfi}: Masquerade destination with no established flow, dropping packet (cannot initiate a connection towards a masquerade expose)" + "{nfi}: Flow origin {} differs from route dst {}. Will drop", + flow_summary.src_vpcd, route.dst_vpcd ); + packet.done(DoneReason::Filtered); + packet.invalidate_flows(); + return; + } + + // check if the flow should be invalidated because it does not have the state required + // to process the flow according to the route requirements + let requirements = Self::route_nat_requirements(route); + if self.should_invalidate_flow(&requirements, input.src_vpcd, Some(flow_summary)) { 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)" - ); - packet.meta_mut().dst_vpcd = Some(dst_vpcd); - Self::set_nat_requirements(packet.meta_mut(), src_nat_mode, dst_nat_mode); + // use the original flow to guide the packet + let flowkey = flow_summary.flow_info.flowkey(); + debug!("{nfi}: Evaluated flow {flowkey} seems valid. Will let packet through"); + Self::tag_for_bypass(packet.meta_mut(), flow_summary); + } + + fn route_packet( + &self, + packet: &mut Packet, + route: &Route, + flow_summary: Option<&FlowSummary>, + ) { + let nfi = &self.name; + debug!("{nfi}: Packet matches route {route}"); + + // Annotate destination and requirements in packet + packet.meta_mut().dst_vpcd = Some(route.dst_vpcd); + Self::set_nat_requirements(packet.meta_mut(), route.src_nat_mode, route.dst_nat_mode); + + // if originator requires port-forwarding and the packet has no active port-forwarding flow, + // drop the packet since port-forwarding should not initiate flows. + if route.src_nat_mode == Some(NatRequirement::PortForwarding) + && !has_active_pfw_flow(flow_summary) + { + debug!("{nfi}: dropping packet without active port-forwarding flow"); + packet.done(DoneReason::Filtered); + packet.invalidate_flows(); + return; + } // Port forwarding or masquerading used in combination with static NAT need to keep track of // the initial IP addresses for creating the right flow table entries, so we may have to @@ -240,17 +340,55 @@ impl FlowFilter { // lacks the NAT context and state to do so. Therefore, it should not upgrade flow to newer // gen ids. However, it can (and must) invalidate flows in some cases, because no other // network function will do it otherwise. - if self.should_invalidate_flow(packet.meta(), dst_vpcd, genid, flow_summary) { + if self.should_invalidate_flow(packet.meta(), route.dst_vpcd, flow_summary) { packet.invalidate_flows(); } } - fn tag_for_bypass( - meta: &mut PacketMeta, - dst_vpcd: VpcDiscriminant, - flow_summary: &FlowSummary, + /// Phase C: apply a resolved route (or drop on a miss) to a single packet. + /// The input `LookupInput` may not correspond to the packet but to the flow + /// in the reverse direction that initiated the flow that this packet matched. + fn apply_route( + &self, + packet: &mut Packet, + flow_summary: Option<&FlowSummary>, + input: LookupInput, + result: LookupResult, ) { - meta.dst_vpcd = Some(dst_vpcd); + let nfi = &self.name; + match result { + LookupResult::Route(route) => { + if let Some(s) = flow_summary.filter(|s| !s.is_master_flow) { + // if the packet did not hit a master (initiating) flow, we queried for the reverse + // flow and the route we get here is not for this packet but for a packet in the reverse + // direction. If we got here, that means that the reverse flow (the initiating) is allowed + // with the current config. But, the fact that such a packet is allowed does not imply that + // the flow the present packet matched is valid. We must check if the flow would be + // compatible with that route in terms of destination and packet treatment (nat mode). + self.validate_flow_from_reverse_route(packet, s, &route, input); + return; + } + // We got a route for the packet, so we know where to send it and how to process it. + self.route_packet(packet, &route, flow_summary); + } + LookupResult::DestinationMiss => { + let dst = input.dst_ip; + debug!("{nfi}: Failed to determine dst VPC for dst {dst}. Will drop"); + packet.invalidate_flows(); + packet.done(DoneReason::Filtered); + } + LookupResult::SourceMiss(dst_vpcd) => { + let src = input.src_ip; + let svpc = input.src_vpcd; + debug!("{nfi}: Source {src} @ {svpc} is not allowed to VPC {dst_vpcd}. Will drop"); + packet.invalidate_flows(); + packet.done(DoneReason::Filtered); + } + } + } + + fn tag_for_bypass(meta: &mut PacketMeta, flow_summary: &FlowSummary) { + meta.dst_vpcd = Some(flow_summary.dst_vpcd); if flow_summary.needs_masquerade { meta.set_masquerade(true); } @@ -294,17 +432,13 @@ impl FlowFilter { &self, meta: &PacketMeta, new_dst_vpcd: VpcDiscriminant, - genid: i64, flow_summary: Option<&FlowSummary>, ) -> bool { let Some(flow_summary) = flow_summary else { return false; }; - if flow_summary.genid == genid { - 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; } @@ -327,39 +461,6 @@ impl FlowFilter { // the flow accordingly). false } - - fn dst_vpcd_from_valid_flow( - &self, - flow_summary: &FlowSummary, - genid: i64, - ) -> Option { - let nfi = &self.name; - if flow_summary.flow_info.status() != FlowStatus::Active { - debug!("{nfi}: Packet has inactive flow information"); - return None; - } - let flow_genid = flow_summary.flow_info.genid(); - if flow_genid < genid && !flow_summary.needs_masquerade { - // If a packet belongs to a masqueraded flow, we have to let it through, temporarily, even if the - // flow is out-dated in terms of generation Id: the flow filter does not have the knowledge to - // forbid that flow. That's the responsibility of the masquerade stage. - debug!( - "{nfi}: Packet has outdated flow information from a prior configuration ({flow_genid} < {genid})" - ); - return None; - } - - let Some(dst_vpcd) = flow_summary.dst_vpcd else { - error!("{nfi}: Flow info does not specify dst VPC. This is a bug. Ignoring it..."); - flow_summary.flow_info.invalidate_pair(); - return None; - }; - - // The flow has the same generation id as the current config (small transient period aside), or - // it is masquerading. So, packet can bypass the flow filter. - debug!("{nfi}: Packet can bypass flow filter thanks to flow information"); - Some(dst_vpcd) - } } impl NetworkFunction for FlowFilter { @@ -382,22 +483,23 @@ impl NetworkFunction for FlowFilter { #[derive(Debug, Clone)] struct FlowSummary { + is_master_flow: bool, genid: i64, - dst_vpcd: Option, + src_vpcd: VpcDiscriminant, + dst_vpcd: VpcDiscriminant, needs_masquerade: bool, needs_port_forwarding: bool, flow_info: Arc, } impl FlowSummary { - fn from_meta(meta: &PacketMeta) -> Option { - let Some(flow_info) = &meta.flow_info else { - return None; - }; + fn from_flow_info(flow_info: &Arc) -> Option { let locked_info = flow_info.locked.read(); Some(Self { + is_master_flow: flow_info.get_flags().is_pair_master(), genid: flow_info.genid(), - dst_vpcd: locked_info.dst_vpcd, + src_vpcd: flow_info.flowkey().src_vpcd()?, + dst_vpcd: locked_info.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(), @@ -405,21 +507,12 @@ 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) - }) +fn has_active_pfw_flow(summary: Option<&FlowSummary>) -> bool { + summary + .filter(|summary| { + summary.flow_info.status() == FlowStatus::Active && summary.needs_port_forwarding + }) + .is_some() } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index 376cc5b524..72e2835d89 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -23,6 +23,7 @@ use net::packet::{DoneReason, Packet, VpcDiscriminant}; use net::parse::DeParse; use pipeline::{NetworkFunction, PipelineData}; use std::time::{Duration, Instant}; +use tracing_test::traced_test; // ------------------------------------------------------------------------------------------------- // Helpers @@ -38,46 +39,140 @@ fn packet(src_vpcd: Option, headers: Headers) -> Packet, + sibling: Arc, +} + +impl AttachedFlow { + // The other half of the pair: the master when the reply flow was attached, and vice versa. + fn sibling(&self) -> &Arc { + &self.sibling + } + + // The attached half as an `Arc`, for tests that need to hand it on rather than query it. + fn arc(&self) -> Arc { + self.attached.clone() + } +} + +impl std::ops::Deref for AttachedFlow { + type Target = FlowInfo; + fn deref(&self) -> &FlowInfo { + &self.attached + } +} + +// Populate one half of a pair the way a downstream stateful NF would: `dst_vpcd` is the flow's +// recorded destination (`None` models a buggy flow with no destination), `nat_state` / +// `port_fw_state` model stored masquerade / port-forwarding state. +fn set_flow_state( + flow: &Arc, + dst_vpcd: Option, + active: bool, + nat_state: bool, + port_fw_state: bool, +) { + if active { + flow.update_status(FlowStatus::Active); + } + let mut locked = flow.locked.write(); + locked.dst_vpcd = dst_vpcd; + if nat_state { + // The concrete type would be a NatState; a bool is enough here since the flow filter + // only checks for presence, never downcasts it. + locked.nat_state = Some(Box::new(true)); + } + if port_fw_state { + locked.port_fw_state = Some(Box::new(true)); + } +} + +// Which direction of a session the attached flow represents -- i.e. whether it is the pair's +// master. `ReplyTo` carries the key the session was opened with, because for a translated session +// that key cannot be derived from the reply's own: the initiator's address has been rewritten, so +// reversing the reply key yields the translated address, not the original one. +#[derive(Clone, Copy)] +enum FlowRole { + /// The packet is the initiating direction, so its own key is the master's. + Initiating, + /// The packet is the reply direction of a session opened with this key. + ReplyTo(FlowKey), +} + +// Attach one half of a flow pair to a packet, the way a downstream stateful NF would (see +// `nat::masquerade::nf::create_flow_pair`: the initiating key is the master, the reverse key is +// not). `dst_vpcd` is the *attached* flow's recorded destination -- for a reply, the initiator's +// VPC. `active` controls the attached flow's status; only active flows can bypass the filter. The +// master is always left Active, so a test that wants otherwise reaches it through `sibling()`. +fn attach_flow_as( packet: &mut Packet, + role: FlowRole, dst_vpcd: Option, active: bool, nat_state: bool, port_fw_state: bool, -) -> Arc { - let flow_key = FlowKey::try_from(&*packet).unwrap(); +) -> AttachedFlow { + let packet_key = FlowKey::try_from(&*packet).unwrap(); + let (master_key, reply_key) = match role { + FlowRole::Initiating => (packet_key, packet_key.reverse(dst_vpcd)), + FlowRole::ReplyTo(master_key) => (master_key, packet_key), + }; let expires_at = Instant::now() + Duration::from_secs(60); - let (flow_info, _) = FlowInfo::related_pair( + let (master, reply) = FlowInfo::related_pair( expires_at, - flow_key, + master_key, packet.meta().compute_flow_flags_forward(), - flow_key.reverse(dst_vpcd), + reply_key, packet.meta().compute_flow_flags_reverse(), ); - if active { - flow_info.update_status(FlowStatus::Active); - } - { - let mut locked = flow_info.locked.write(); - locked.dst_vpcd = dst_vpcd; - if nat_state { - // The concrete type would be a NatState; a bool is enough here since the flow filter - // only checks for presence, never downcasts it. - locked.nat_state = Some(Box::new(true)); - } - if port_fw_state { - locked.port_fw_state = Some(Box::new(true)); + let (attached, sibling) = match role { + FlowRole::Initiating => (master, reply), + FlowRole::ReplyTo(_) => { + // The master's own destination is the VPC this reply came from. + set_flow_state( + &master, + packet.meta().src_vpcd, + true, + nat_state, + port_fw_state, + ); + (reply, master) } - } - packet.meta_mut().flow_info = Some(flow_info.clone()); - flow_info + }; + + set_flow_state(&attached, dst_vpcd, active, nat_state, port_fw_state); + packet.meta_mut().flow_info = Some(attached.clone()); + AttachedFlow { attached, sibling } +} + +// Attach the initiating (master) half, built from the packet's own key: the common case. +fn attach_flow( + packet: &mut Packet, + dst_vpcd: Option, + active: bool, + nat_state: bool, + port_fw_state: bool, +) -> AttachedFlow { + attach_flow_as( + packet, + FlowRole::Initiating, + dst_vpcd, + active, + nat_state, + port_fw_state, + ) +} + +// The key a session was opened with, for `FlowRole::ReplyTo`: the initiating packet's own key. +fn initiating_key(src_vpcd: Option, headers: Headers) -> FlowKey { + FlowKey::try_from(&packet(src_vpcd, headers)).unwrap() } fn make_flow_filter(ctx: FlowFilterContext) -> (FlowFilter, FlowFilterContextWriter) { @@ -182,9 +277,16 @@ fn ipv6_context() -> FlowFilterContext { // ------------------------------------------------------------------------------------------------- // Basic acceptance / rejection +fn show_flow_filter(flow_filter: &FlowFilter) { + let tables = flow_filter.tables.load(); + println!("{tables}"); +} + #[test] +#[traced_test] fn allowed_packet_sets_destination_and_no_nat() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); + show_flow_filter(&flow_filter); let out = run( &mut flow_filter, packet( @@ -200,8 +302,10 @@ fn allowed_packet_sets_destination_and_no_nat() { } #[test] +#[traced_test] fn unmatched_destination_is_filtered() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); + show_flow_filter(&flow_filter); let out = run( &mut flow_filter, packet( @@ -214,6 +318,7 @@ fn unmatched_destination_is_filtered() { } #[test] +#[traced_test] fn missing_source_vpc_is_unroutable() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); let out = run( @@ -287,6 +392,7 @@ fn static_nat_source_sets_static_flag() { } #[test] +#[traced_test] fn masquerade_source_sets_masquerade_flag() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); let out = run( @@ -303,6 +409,7 @@ fn masquerade_source_sets_masquerade_flag() { } #[test] +#[traced_test] fn port_forwarding_destination_sets_flag_and_is_protocol_aware() { let (mut flow_filter, _) = make_flow_filter(dst_port_forwarding_context()); @@ -333,6 +440,7 @@ fn port_forwarding_destination_sets_flag_and_is_protocol_aware() { // Stateful flows #[test] +#[traced_test] fn active_flow_state_is_honored() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); // Route itself requires no NAT, but the attached active flow carries masquerade state, so the @@ -351,6 +459,7 @@ fn active_flow_state_is_honored() { } #[test] +#[traced_test] fn outdated_flow_is_invalidated() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); // Advance the configuration generation so the flow (genid 0) is outdated. @@ -449,6 +558,7 @@ fn batch_of_packets_is_processed_independently() { // Stateful flows: bypass eligibility #[test] +#[traced_test] fn active_flow_port_forwarding_state_is_honored() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); // No-NAT route, but the active flow carries port-forwarding state -> tagged for port forwarding. @@ -465,6 +575,7 @@ fn active_flow_port_forwarding_state_is_honored() { } #[test] +#[traced_test] fn inactive_flow_state_is_not_honored() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); // The flow carries masquerade state but is not active, so it must not be used to bypass the @@ -514,6 +625,7 @@ fn outdated_flow_that_no_longer_needs_state_is_invalidated() { assert_eq!(flow.status(), FlowStatus::Cancelled); } +#[traced_test] #[test] fn outdated_flow_missing_masquerade_state_is_invalidated() { // Outdated flow, correct destination, route now requires masquerade, but the flow has no @@ -525,6 +637,7 @@ fn outdated_flow_missing_masquerade_state_is_invalidated() { build_tcp_packet(v4("3.0.0.5"), v4("5.0.0.10"), 1234, 5678), ); let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, false); + flow.set_genid(4); let out = run(&mut flow_filter, p); assert!(!out.is_done(), "{:?}", out.get_done()); assert!(out.meta().requires_masquerade()); @@ -532,6 +645,7 @@ fn outdated_flow_missing_masquerade_state_is_invalidated() { } #[test] +#[traced_test] fn outdated_flow_missing_port_forwarding_state_is_invalidated() { // Outdated flow, correct destination, route now requires port forwarding, but the flow has no // port-forwarding state. @@ -542,13 +656,14 @@ fn outdated_flow_missing_port_forwarding_state_is_invalidated() { build_tcp_packet(v4("10.0.0.5"), v4("80.0.0.5"), 1234, 2222), ); let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, false); + flow.set_genid(4); let out = run(&mut flow_filter, p); assert!(!out.is_done(), "{:?}", out.get_done()); assert!(out.meta().requires_port_forwarding()); - assert_eq!(flow.status(), FlowStatus::Cancelled); } #[test] +#[traced_test] fn outdated_flow_with_consistent_state_is_kept() { // Outdated flow, correct destination, route requires masquerade and the flow already has // masquerade state: the filter cannot prove it stale, so it is left for the stateful NFs. @@ -566,24 +681,50 @@ 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. +// +// A reply packet's own 5-tuple is unanswerable by the tables: its source is a translated address +// and its destination is a masquerade public range, which is withheld from the remote tables +// altogether. So when a reply-direction (non-master) flow falls behind the generation id, the +// filter re-validates the session by looking up the *initiating* direction -- the master flow's +// key -- and then checks that the answer still describes this session. +// +// The pair is built the way the masquerade NF builds it: initiating key is the master, reverse key +// is not (`FlowRole::ReplyTo`). + +// The initiating direction of the masqueraded session used below: vpc1's host 3.0.0.5 (inside the +// 3.0.0.0/24 masquerade expose) opening a connection to vpc2's 5.0.0.10. +fn masquerade_session_key() -> FlowKey { + initiating_key( + Some(vpcd(100)), + build_tcp_packet(v4("3.0.0.5"), v4("5.0.0.10"), 1234, 5678), + ) +} + +// A reply packet for that session: vpc2 answering towards vpc1's masquerade public address. +fn masquerade_reply_packet() -> Packet { + packet( + Some(vpcd(200)), + build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), + ) +} #[test] fn masquerade_reply_on_established_flow_survives_config_change() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); set_genid(&mut flow_filter, 5); - // Reply direction of a masqueraded session: vpc2 answers towards vpc1's masquerade public - // range. The flow (genid 0) is outdated, so the bypass is refused and the packet goes through - // the tables, which resolve a masquerade marker. - let mut p = packet( - Some(vpcd(200)), - build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), + // The flow (genid 0) is outdated, so the bypass is refused; the initiating direction still + // routes to vpc2 with masquerade on the source, which is what this flow records. + let mut p = masquerade_reply_packet(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masquerade_session_key()), + Some(vpcd(100)), + true, + true, + false, ); - 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))); @@ -591,21 +732,111 @@ fn masquerade_reply_on_established_flow_survives_config_change() { assert_ne!(flow.status(), FlowStatus::Cancelled); } +// The initiating direction still routes, but the session is no longer the same kind of session: +// the flow carries no masquerade state while the route now calls for it. A reply cannot be +// re-annotated from the route the way a forward packet can (its addresses are already +// translated), so the session ends. +#[test] +fn masquerade_reply_is_dropped_when_the_nat_requirement_changed() { + let (mut flow_filter, _) = make_flow_filter(source_nat_context()); + set_genid(&mut flow_filter, 5); + let mut p = masquerade_reply_packet(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masquerade_session_key()), + Some(vpcd(100)), + true, + false, // no masquerade state, but the route for the initiating direction requires it + false, + ); + + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + +// The value the packet would be forwarded on -- the reply flow's recorded destination -- has to be +// the initiator's VPC, which is the master key's source VPC. A pair that disagrees is malformed. +#[test] +fn masquerade_reply_with_wrong_initiator_vpc_is_dropped() { + let (mut flow_filter, _) = make_flow_filter(source_nat_context()); + set_genid(&mut flow_filter, 5); + let mut p = masquerade_reply_packet(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masquerade_session_key()), // opened from vpc1... + Some(vpcd(300)), // ...but the reply claims to be headed to vpc3 + true, + true, + false, + ); + + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + +// Without a live master flow there is no initiating direction to validate against, so there is no +// question the tables can be asked. Fail closed rather than admit the packet un-annotated, and +// tear the pair down: nothing will ever be able to validate it again, so leaving it would drop one +// packet per arrival until it aged out. +#[test] +fn masquerade_reply_without_a_live_master_flow_is_dropped() { + let (mut flow_filter, _) = make_flow_filter(source_nat_context()); + set_genid(&mut flow_filter, 5); + let mut p = masquerade_reply_packet(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masquerade_session_key()), + Some(vpcd(100)), + true, + true, + false, + ); + flow.sibling().update_status(FlowStatus::Expired); + + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Unroutable)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + +// The peering is gone from the new config, so the initiating direction no longer resolves at all: +// the session is over and the flow pair goes with it. +#[test] +fn masquerade_reply_does_not_survive_peering_removal() { + let (mut flow_filter, writer) = make_flow_filter(source_nat_context()); + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, 5); + let mut p = masquerade_reply_packet(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masquerade_session_key()), + 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); +} + #[test] +#[traced_test] fn masquerade_reply_without_flow_is_filtered() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); - // No established flow: a masquerade destination cannot accept a new connection. - let out = run( - &mut flow_filter, - packet( - Some(vpcd(200)), - build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), - ), + let packet = packet( + Some(vpcd(200)), + build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), ); + // no flow with masquerade, packet is droppped + let out = run(&mut flow_filter, packet); assert_eq!(out.get_done(), Some(DoneReason::Filtered)); } #[test] +#[traced_test] fn masquerade_reply_with_inactive_flow_is_filtered() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); set_genid(&mut flow_filter, 5); @@ -613,12 +844,15 @@ fn masquerade_reply_with_inactive_flow_is_filtered() { Some(vpcd(200)), build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), ); - attach_flow(&mut p, Some(vpcd(100)), false, true, false); + let flow_info = attach_flow(&mut p, Some(vpcd(100)), false, true, false); + flow_info.set_genid(5); + flow_info.update_status(FlowStatus::Expired); let out = run(&mut flow_filter, p); assert_eq!(out.get_done(), Some(DoneReason::Filtered)); } #[test] +#[traced_test] fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { let (mut flow_filter, _) = make_flow_filter(source_nat_context()); set_genid(&mut flow_filter, 5); @@ -628,15 +862,20 @@ fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { ); // The packet hits a flow that is masquerading. The flow-filter will not set the verdict but be bypassed. let flow = attach_flow(&mut p, Some(vpcd(300)), true, true, false); + flow.set_genid(5); let out = run(&mut flow_filter, p); assert_eq!(out.get_done(), None); assert_eq!(flow.status(), FlowStatus::Active); } #[test] +#[traced_test] 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); + let table = flow_filter.tables.load(); + println!("{table}"); + // Reply direction of a forwarded session: the forwarded host answers from its private // address, which is (deliberately) not in the local tables. let mut p = packet( @@ -644,6 +883,7 @@ fn port_forwarding_reply_on_established_flow_survives_config_change() { build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), ); let flow = attach_flow(&mut p, Some(vpcd(100)), true, false, true); + flow.set_genid(4); let out = run(&mut flow_filter, p); assert!(!out.is_done(), "{:?}", out.get_done()); assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); @@ -664,22 +904,9 @@ fn port_forwarding_reply_without_flow_is_filtered() { assert_eq!(out.get_done(), Some(DoneReason::Filtered)); } -#[test] -fn masquerade_flow_is_left_untouched_on_config_removal() { - // The peering is gone from the new config: even an active, state-consistent flow must not let - // reply traffic through (stage 1 finds no marker to trust), and the flow pair is invalidated. - let (mut flow_filter, writer) = make_flow_filter(source_nat_context()); - writer.store(context(&[], vec![])); - set_genid(&mut flow_filter, 5); - let mut p = packet( - Some(vpcd(200)), - build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), - ); - let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); - let out = run(&mut flow_filter, p); - assert_eq!(out.get_done(), None); - assert_eq!(flow.status(), FlowStatus::Active); -} +// (peering removal for a masquerade session is covered by +// `masquerade_reply_does_not_survive_peering_removal`, which models the reply flow as the +// non-master half the masquerade NF actually creates) #[test] fn portfw_flow_does_not_survive_peering_removal() { @@ -865,36 +1092,29 @@ fn mixed_v4_v6_burst_partitions_by_version_and_preserves_order() { // subtlest policy in this NF (it decides which established sessions a config change kills); the // property pins its complete truth table instead of sampling branches. -#[derive(Debug, Clone, Copy, bolero::TypeGenerator)] -enum GenidRel { - Older, - Same, - Newer, -} - #[derive(Debug, Clone, Copy, bolero::TypeGenerator)] struct InvalidationCase { meta_masquerade: bool, 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, + /// Whether the flow's recorded destination equals the route's. A flow with no destination at + /// all is not representable: `FlowSummary` only exists for flows that record one. + 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. +// The specification: a flow is invalidated iff the filter can prove it stale: the 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. The +// generation is NOT part of this decision: `classify` only routes flows that already need +// re-validation (an older genid) down this path, so the function trusts its caller on that point +// -- see the precondition on `should_invalidate_flow`. fn expected_invalidation(case: &InvalidationCase) -> bool { - if !case.has_flow || matches!(case.genid, GenidRel::Same) { + if !case.has_flow { 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) @@ -926,23 +1146,24 @@ fn invalidation_decision_matches_spec() { meta.set_port_forwarding(case.meta_port_forwarding); let summary = case.has_flow.then(|| crate::FlowSummary { - genid: match case.genid { - GenidRel::Older => GENID - 3, - 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, + is_master_flow: true, + // Older than the current generation: the only case that reaches the function. + genid: GENID - 3, + // The packet's own VPC. Not part of this decision; only the reverse-route path + // reads it. + src_vpcd: vpcd(100), + dst_vpcd: if case.flow_dst_matches { + route_dst + } else { + vpcd(300) }, needs_masquerade: case.flow_masquerade, needs_port_forwarding: case.flow_port_forwarding, - flow_info, + flow_info: flow_info.arc(), }); assert_eq!( - flow_filter.should_invalidate_flow(&meta, route_dst, GENID, summary.as_ref()), + flow_filter.should_invalidate_flow(&meta, route_dst, summary.as_ref()), expected_invalidation(case), "decision diverges from spec for {case:?}", ); @@ -1069,10 +1290,20 @@ fn burst_processing_upholds_structural_invariants() { "processed packet must be resolved XOR done: {spec:?}", ); - // An active, current-generation flow with a recorded destination always - // short-circuits the tables, whatever they would have said. + // A packet is disqualified before its flow is ever consulted: no IP headers, or no + // source VPC, and it is dropped whatever flow it carries. An attached flow does + // not excuse a packet the filter cannot form a lookup key for. + if spec.non_ip { + assert_eq!(pkt.get_done(), Some(DoneReason::NotIp), "{spec:?}"); + } else if !spec.has_src_vpcd { + assert_eq!(pkt.get_done(), Some(DoneReason::Unroutable), "{spec:?}"); + } + + // Having cleared that, an active, current-generation flow with a recorded + // destination always short-circuits the tables, whatever they would have said. if let Some(f) = spec.flow && !spec.non_ip + && spec.has_src_vpcd && !bump_genid && f.active && f.dst.is_some() @@ -1081,7 +1312,11 @@ fn burst_processing_upholds_structural_invariants() { assert_eq!(pkt.meta().dst_vpcd, f.dst_vpcd(), "{spec:?}"); } - // A Filtered packet always cancels its flow pair (Unroutable/NotIp do not). + // A Filtered packet always cancels its flow pair. The disqualification drops above + // (NotIp, and the Unroutable for a missing source VPC) leave the flow alone -- + // they say nothing about whether the session is still valid. The one Unroutable + // that does cancel is the missing-master case, which no spec here can generate: + // `attach_flow` only ever attaches the master half. if pkt.get_done() == Some(DoneReason::Filtered) && let Some(flow) = &flows[i] { From 87796b0a4e9ad805be57cf920ce5485cd99d62a8 Mon Sep 17 00:00:00 2001 From: Fredi Raspall Date: Thu, 6 Aug 2026 19:55:02 +0200 Subject: [PATCH 6/6] feat(flow-filter): add tests (Claude) Signed-off-by: Fredi Raspall --- flow-filter/src/tests.rs | 338 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 338 insertions(+) diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index 72e2835d89..a39b5b0c0d 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -936,6 +936,344 @@ fn portfw_flow_does_not_survive_peering_removal() { assert_eq!(flow.status(), FlowStatus::Cancelled); } +// ------------------------------------------------------------------------------------------------- +// Flow re-validation when exposed prefixes overlap. +// +// Ported from `pr/qmonnet/flow-filter-table` (`revalidation_works_in_case_of_*_overlap`), which +// gates the un-initiable rules on flow-supplied key material so that a reply's own 5-tuple can be +// resolved. This NF answers the same scenarios from the other end: a reply is re-validated by +// looking up the key the session was *opened* with, so the pairs below are built the way the +// stateful NFs build them (initiating direction is the master, reply direction is not). The +// overlaps are then resolved by which session a packet belongs to rather than by its own tuple -- +// the assertions are the other branch's, unchanged. + +#[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); + + // The session: vpc2's 2.0.0.1 opens a connection to vpc1's 1.0.0.1, masqueraded behind + // 10.0.0.1. vpc3 masquerades the very same prefix pair, so the reply's own destination + // (10.0.0.1) does not say which VPC the session belongs to -- the master key does. + let session = initiating_key( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let request = || { + packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ) + }; + let reply = || { + packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ) + }; + + // Initial packet from vpc2 to vpc1 (no flow info) passes + let out = run(&mut flow_filter, request()); + 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 = reply(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(session), + 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 = request(); + 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 = reply(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(session), + 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 outdated flow info) passes + let mut p = request(); + 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); + + // Remove peerings, 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 = reply(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(session), + 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); + + // Two sessions over the overlapping prefixes: one forwarded, opened from vpc2 towards vpc1's + // public 10.0.0.1:5000 (forwarded to 1.0.0.1:2000), and one masqueraded, opened from vpc1's + // 1.0.0.1:2000 towards vpc2 behind that same public socket. Each session's reply carries the + // other session's initiating 5-tuple, so the tuple alone cannot say which session a packet + // belongs to -- hence two builders, used in both roles below. + let forwarded_session = initiating_key( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let masqueraded_session = initiating_key( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + // vpc2 -> vpc1's public socket: the forwarded session's request, the masqueraded one's reply. + let vpc2_to_public = || { + packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ) + }; + // vpc1 -> vpc2: the masqueraded session's request, the forwarded one's reply. + let vpc1_to_vpc2 = || { + packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ) + }; + + // Port-forwarding: Initial packet from vpc2 to vpc1 (no flow info) passes + let out = run(&mut flow_filter, vpc2_to_public()); + 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 = vpc1_to_vpc2(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(forwarded_session), + 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 = vpc2_to_public(); + 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 out = run(&mut flow_filter, vpc1_to_vpc2()); + 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 = vpc2_to_public(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masqueraded_session), + 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 = vpc1_to_vpc2(); + 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 = vpc1_to_vpc2(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(forwarded_session), + 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 = vpc2_to_public(); + 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 = vpc2_to_public(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masqueraded_session), + 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 = vpc1_to_vpc2(); + 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); + + // Remove 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 = vpc1_to_vpc2(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(forwarded_session), + 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 = vpc2_to_public(); + let flow = attach_flow_as( + &mut p, + FlowRole::ReplyTo(masqueraded_session), + 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