diff --git a/Cargo.lock b/Cargo.lock index 59f514d47e..c818995741 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1454,6 +1454,7 @@ dependencies = [ "indenter", "linkme", "tracing", + "tracing-test", ] [[package]] diff --git a/flow-filter/Cargo.toml b/flow-filter/Cargo.toml index 627c93494c..8ebae286b9 100644 --- a/flow-filter/Cargo.toml +++ b/flow-filter/Cargo.toml @@ -34,3 +34,4 @@ dpdk = { workspace = true, features = ["test"] } lpm = { workspace = true, features = ["testing"] } # Enable generated header stacks for classifier tests. net = { workspace = true, features = ["builder", "bolero"] } +tracing-test = { workspace = true } diff --git a/flow-filter/src/context/display.rs b/flow-filter/src/context/display.rs index b9561c9bc6..267bc3d74d 100644 --- a/flow-filter/src/context/display.rs +++ b/flow-filter/src/context/display.rs @@ -6,7 +6,7 @@ //! Tables retain typed rules for backend-independent display. Each field's type controls its //! formatting, keeping values coupled to their key fields. -use super::tables::FlowFilterContext; +use super::tables::{FlowFilterContext, GateVni, SourceGate}; impl crate::NatRequirement { fn label(self) -> &'static str { @@ -24,6 +24,24 @@ impl std::fmt::Display for crate::NatRequirement { } } +impl std::fmt::Display for GateVni { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self.0 { + Some(vni) => write!(f, "{vni}"), + None => f.write_str("-"), + } + } +} + +impl std::fmt::Display for SourceGate { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SourceGate::Ungated => f.write_str("-"), + SourceGate::PortFwdReply => f.write_str("pfwd"), + } + } +} + // ------------------------------------------------------------------------------------------------- // Rendering: one section per table, each rule on a line, in match order. diff --git a/flow-filter/src/context/fuzz.rs b/flow-filter/src/context/fuzz.rs index 27eee723a2..8f920d2ed6 100644 --- a/flow-filter/src/context/fuzz.rs +++ b/flow-filter/src/context/fuzz.rs @@ -11,12 +11,13 @@ #![cfg(test)] -use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult}; +use super::tables::{Backend, FlowFilterContext, LookupInput, LookupResult, SourceGate}; use crate::NatRequirement; use crate::fuzz_gen::{OverlaySpec, Probe, ProbeSpec, bogus_vpcd}; use concurrency::sync::LazyLock; use concurrency::sync::atomic::{AtomicU64, Ordering}; use config::external::overlay::ValidatedOverlay; +use config::external::overlay::vpc::{ValidatedPeering, ValidatedVpc}; use lpm::prefix::{IpPrefix, L4Protocol, Prefix, PrefixWithOptionalPorts}; use net::ip::NextHeader; use net::packet::VpcDiscriminant; @@ -66,28 +67,31 @@ fn consider(best: &mut Option<(Precedence, T)>, precedence: Precedence, value } } -/// Answer a route lookup directly from the validated overlay. -/// Shared by the context and NF metadata property tests. -pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { - let Some(src_vpc) = overlay - .vpc_table() - .values() - .find(|vpc| VpcDiscriminant::from_vni(vpc.vni()) == probe.src_vpcd) - else { - return LookupResult::DestinationMiss; - }; - if probe.src_ip.is_ipv4() != probe.dst_ip.is_ipv4() { - return LookupResult::DestinationMiss; - } - let (sport, dport) = probe.ports.unwrap_or((0, 0)); - - // Stage 1: the destination against every peer's public prefixes. Masquerade exposes are - // included (marker rules); a default expose acts as a /0 of the peering's IP version. +/// Stage 1: the destination against every peer's public prefixes, scoped to the source VPC. +/// +/// `revalidated` is the destination VPC an outdated flow vouches for. With it, only the masquerade +/// exposes of the peering to that VPC answer -- they cannot receive connections, so nothing but a +/// flow that already used them can reach them. Without it, only exposes that can receive do (a +/// default expose acts as a /0 of the peering's IP version). +fn oracle_stage1( + src_vpc: &ValidatedVpc, + probe: &Probe, + dport: u16, + revalidated: Option, +) -> Option<(VpcDiscriminant, Option)> { let mut verdict: Option<(Precedence, (VpcDiscriminant, Option))> = None; for peering in src_vpc.peerings() { let dst_vpcd = VpcDiscriminant::from_vni(peering.remote_vni()); + if revalidated.is_some_and(|vpcd| vpcd != dst_vpcd) { + continue; + } for expose in peering.remote().valexp() { - if !proto_allows(expose.nat_proto(), probe.proto) { + let matchable = if revalidated.is_some() { + expose.has_masquerade() + } else { + expose.can_receive_connection() + }; + if !matchable || !proto_allows(expose.nat_proto(), probe.proto) { continue; } for prefix in expose.public_ips() { @@ -100,29 +104,35 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup } } } - if peering.remote().has_default_expose() && probe.dst_ip.is_ipv4() == peering.is_v4() { + if revalidated.is_none() + && peering.remote().has_default_expose() + && probe.dst_ip.is_ipv4() == peering.is_v4() + { consider(&mut verdict, (0, false), (dst_vpcd, None)); } } - let Some((_, (dst_vpcd, dst_nat))) = verdict else { - return LookupResult::DestinationMiss; - }; + verdict.map(|(_, hit)| hit) +} - // Stage 2: the source against that peering's private prefixes. Port-forwarding sources are - // excluded (they cannot initiate); a default expose acts as a /0 of the peering's version. - let peering = src_vpc - .peerings() - .iter() - .find(|p| VpcDiscriminant::from_vni(p.remote_vni()) == dst_vpcd) - .unwrap_or_else(|| unreachable!("stage 1 hit implies a peering to the verdict VPC")); +/// Stage 2: the source against the resolved peering's private prefixes. +/// +/// `gate` is what the lookup asks about. Gated on port forwarding, only the port-forwarding +/// exposes answer -- they cannot initiate connections, so nothing but a flow that already used +/// them can reach them. Ungated, only exposes that can initiate do (a default expose acts as a /0 +/// of the peering's IP version). +fn oracle_stage2( + peering: &ValidatedPeering, + probe: &Probe, + sport: u16, + gate: SourceGate, +) -> Option> { let mut src_nat: Option<(Precedence, Option)> = None; - for expose in peering - .local() - .valexp() - .iter() - .filter(|expose| expose.can_init_connection()) - { - if !proto_allows(expose.nat_proto(), probe.proto) { + for expose in peering.local().valexp() { + let matchable = match gate { + SourceGate::PortFwdReply => expose.has_port_forwarding(), + SourceGate::Ungated => expose.can_init_connection(), + }; + if !matchable || !proto_allows(expose.nat_proto(), probe.proto) { continue; } for prefix in expose.ips() { @@ -135,11 +145,52 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup } } } - if peering.local().has_default_expose() && probe.src_ip.is_ipv4() == peering.is_v4() { + if !gate.is_gated() + && peering.local().has_default_expose() + && probe.src_ip.is_ipv4() == peering.is_v4() + { consider(&mut src_nat, (0, false), None); } + src_nat.map(|(_, nat)| nat) +} + +/// Answer a route lookup directly from the validated overlay. +/// Shared by the context and NF metadata property tests. +/// +/// Both stages ask what the flow vouches for first, and only then the plain question. A packet on +/// an established flow keeps the peering and the NAT mode that flow was built on, even where an +/// expose covers the same address ungated; the plain question is the fallback, for the forward +/// traffic that carries no flow information of its own. +pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> LookupResult { + let Some(src_vpc) = overlay + .vpc_table() + .values() + .find(|vpc| VpcDiscriminant::from_vni(vpc.vni()) == probe.src_vpcd) + else { + return LookupResult::DestinationMiss; + }; + if probe.src_ip.is_ipv4() != probe.dst_ip.is_ipv4() { + return LookupResult::DestinationMiss; + } + let (sport, dport) = probe.ports.unwrap_or((0, 0)); + + let verdict = probe + .dst_vpcd + .and_then(|vpcd| oracle_stage1(src_vpc, probe, dport, Some(vpcd))) + .or_else(|| oracle_stage1(src_vpc, probe, dport, None)); + let Some((dst_vpcd, dst_nat)) = verdict else { + return LookupResult::DestinationMiss; + }; + + let peering = src_vpc + .peerings() + .iter() + .find(|p| VpcDiscriminant::from_vni(p.remote_vni()) == dst_vpcd) + .unwrap_or_else(|| unreachable!("stage 1 hit implies a peering to the verdict VPC")); + let src_nat = oracle_stage2(peering, probe, sport, probe.gate) + .or_else(|| oracle_stage2(peering, probe, sport, SourceGate::Ungated)); match src_nat { - Some((_, src_nat)) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), + Some(src_nat) => LookupResult::Route((dst_vpcd, dst_nat, src_nat)), None => LookupResult::SourceMiss(dst_vpcd), } } @@ -147,6 +198,32 @@ pub(crate) fn oracle_lookup(overlay: &ValidatedOverlay, probe: &Probe) -> Lookup // ------------------------------------------------------------------------------------------------- // Properties. +/// A probe's fields are exactly one lookup's arguments, revalidation information included. +fn lookup(tables: &FlowFilterContext, probe: &Probe) -> LookupResult { + tables.lookup( + probe.src_vpcd, + probe.dst_vpcd, + probe.src_ip, + probe.dst_ip, + probe.proto, + probe.ports, + probe.gate, + ) +} + +/// The same question, in the batch path's form. +fn lookup_input(probe: &Probe) -> LookupInput { + LookupInput { + src_vpcd: probe.src_vpcd, + dst_vpcd: probe.dst_vpcd, + src_ip: probe.src_ip, + dst_ip: probe.dst_ip, + proto: probe.proto, + ports: probe.ports, + gate: probe.gate, + } +} + /// The rte_acl backend agrees with the reference backend on every probe of every generated /// overlay -- single lookups and the chunked batch path alike. This is the fuzz form of /// `tests::reference_and_dpdk_backends_agree`: it validates the wide-key encoding and the @@ -160,6 +237,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { // Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not // const (each instance registers with the loom executor). static ROUTES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static REVALIDATED_ROUTES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static SOURCE_MISSES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static DESTINATION_MISSES: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -171,68 +249,34 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { .expect("reference build"); let dpdk = FlowFilterContext::build(&built.overlay, Backend::Dpdk).expect("dpdk build"); - // Guaranteed-routing probes derived from the overlay's own structure. A full route from - // random generation would be a rare random outcome (~1% of generated routes), and we - // may not get enough of them during a time-boxed bolero run on a loaded CI runner to - // get meaningful coverage. + // Probes derived from the overlay's own structure route by construction. A full route + // from random generation would be a rare random outcome (~1% of generated routes), and + // we may not get enough of them during a time-boxed bolero run on a loaded CI runner to + // get meaningful coverage -- the revalidated ones would get none at all. for probe in &built.routing_probes { - let want = reference.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - ); - assert_eq!( - dpdk.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports - ), - want, - "backends disagree on derived routing probe {probe:?}\nspec: {overlay_spec:?}", - ); assert!( - matches!(want, LookupResult::Route(_)), + matches!(lookup(&reference, probe), LookupResult::Route(_)), "derived routing probe did not route: {probe:?}\nspec: {overlay_spec:?}", ); - ROUTES.fetch_add(1, Ordering::Relaxed); + if probe.dst_vpcd.is_some() || probe.gate.is_gated() { + REVALIDATED_ROUTES.fetch_add(1, Ordering::Relaxed); + } } - let probes: Vec = probe_specs + // Derived and random probes alike go through both backends and both lookup paths. + let probes: Vec = built + .routing_probes .iter() - .map(|p| p.resolve(built.blocks)) - .collect(); - let inputs: Vec = probes - .iter() - .map(|p| LookupInput { - src_vpcd: p.src_vpcd, - src_ip: p.src_ip, - dst_ip: p.dst_ip, - proto: p.proto, - ports: p.ports, - }) + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))) .collect(); + let inputs: Vec = probes.iter().map(lookup_input).collect(); let mut expected = Vec::with_capacity(probes.len()); for probe in &probes { - let want = reference.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports, - ); + let want = lookup(&reference, probe); assert_eq!( - dpdk.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports - ), + lookup(&dpdk, probe), want, "backends disagree on single lookup of {probe:?}\nspec: {overlay_spec:?}", ); @@ -246,7 +290,7 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { expected.push(want); } - // Batch path: 40 inputs > MAX_BATCH exercises the chunked scatter; every slot must + // Batch path: more than MAX_BATCH inputs exercise the chunked scatter; every slot must // equal the corresponding single lookup. let mut out = vec![LookupResult::DestinationMiss; inputs.len()]; dpdk.lookup_batch(&inputs, &mut out); @@ -261,10 +305,12 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { let all_miss: Vec = (0..33u8) .map(|i| LookupInput { src_vpcd: bogus_vpcd(), + dst_vpcd: None, src_ip: format!("10.0.0.{i}").parse().unwrap(), dst_ip: "10.0.0.99".parse().unwrap(), proto: NextHeader::TCP, ports: Some((1, 2)), + gate: SourceGate::Ungated, }) .collect(); let mut out = vec![LookupResult::DestinationMiss; all_miss.len()]; @@ -276,12 +322,17 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { }); eprintln!( - "coverage: {} routes, {} source misses, {} destination misses", + "coverage: {} routes ({} revalidated), {} source misses, {} destination misses", ROUTES.load(Ordering::Relaxed), + REVALIDATED_ROUTES.load(Ordering::Relaxed), SOURCE_MISSES.load(Ordering::Relaxed), DESTINATION_MISSES.load(Ordering::Relaxed), ); assert!(ROUTES.load(Ordering::Relaxed) >= 1, "no full routes at all"); + assert!( + REVALIDATED_ROUTES.load(Ordering::Relaxed) >= 1, + "no route resolved through revalidation information", + ); assert!( SOURCE_MISSES.load(Ordering::Relaxed) >= 4, "too few source misses" @@ -293,10 +344,11 @@ fn dpdk_backend_matches_reference_on_generated_overlays() { } /// The batched lookup equals the single lookup, slot for slot, on the reference backend. The -/// batch path's own logic -- the v4/v6 partition, `MAX_BATCH` chunking, the stage-1-hit gather -/// and the scatter back through saved indices -- is backend-generic, so this EAL-free variant -/// exercises it with far more iterations than the rte_acl differential can afford. 40 probes -/// force multi-chunk batches; probe specs freely mix IP versions and version-mismatched pairs. +/// batch path's own logic -- the v4/v6 partition, `MAX_BATCH` chunking, the stage-1-hit gather, +/// the revalidation re-lookups and the scatter back through saved indices -- is backend-generic, +/// so this EAL-free variant exercises it with far more iterations than the rte_acl differential +/// can afford. 40 probes force multi-chunk batches; probe specs freely mix IP versions and +/// version-mismatched pairs, and the overlay's derived probes bring the revalidated routes. #[test] fn batched_lookup_matches_single_lookup() { bolero::check!() @@ -306,33 +358,20 @@ fn batched_lookup_matches_single_lookup() { let tables = FlowFilterContext::build(&built.overlay, Backend::Reference) .expect("reference build"); - let probes: Vec = probe_specs + let probes: Vec = built + .routing_probes .iter() - .map(|p| p.resolve(built.blocks)) - .collect(); - let inputs: Vec = probes - .iter() - .map(|p| LookupInput { - src_vpcd: p.src_vpcd, - src_ip: p.src_ip, - dst_ip: p.dst_ip, - proto: p.proto, - ports: p.ports, - }) + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))) .collect(); + let inputs: Vec = probes.iter().map(lookup_input).collect(); let mut out = vec![LookupResult::DestinationMiss; inputs.len()]; tables.lookup_batch(&inputs, &mut out); for (i, probe) in probes.iter().enumerate() { assert_eq!( out[i], - tables.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports - ), + lookup(&tables, probe), "batch slot {i} != single lookup for {probe:?}\nspec: {overlay_spec:?}", ); } @@ -349,16 +388,14 @@ fn reference_lookup_matches_config_oracle() { let built = overlay_spec.build(); let tables = FlowFilterContext::build(&built.overlay, Backend::Reference).expect("reference build"); - for probe_spec in probe_specs { - let probe = probe_spec.resolve(built.blocks); + let probes = built + .routing_probes + .iter() + .copied() + .chain(probe_specs.iter().map(|p| p.resolve(built.blocks))); + for probe in probes { assert_eq!( - tables.lookup( - probe.src_vpcd, - probe.src_ip, - probe.dst_ip, - probe.proto, - probe.ports - ), + lookup(&tables, &probe), oracle_lookup(&built.overlay, &probe), "reference tables disagree with the config oracle on {probe:?}\nspec: {overlay_spec:?}", ); diff --git a/flow-filter/src/context/mod.rs b/flow-filter/src/context/mod.rs index cd703a8403..4d1cb245fe 100644 --- a/flow-filter/src/context/mod.rs +++ b/flow-filter/src/context/mod.rs @@ -17,7 +17,7 @@ mod tests; pub use tables::FlowFilterContext; use tables::PRODUCTION_BACKEND; -pub(crate) use tables::{LookupInput, LookupResult}; +pub(crate) use tables::{LookupInput, LookupResult, SourceGate}; impl TryFrom<&ValidatedOverlay> for FlowFilterContext { type Error = ConfigError; diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index d8e98fabf5..b6a2b4113e 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -18,13 +18,18 @@ //! longest-prefix-match (encoded in the rule priority, see [`rule_priority`]) //! handles it uniformly. //! -//! Masquerade destinations are kept in the remote tables even though they cannot -//! accept new connections: their [`Verdict`] marks reply traffic on established -//! masquerade flows as distinguishable from a destination no peering covers, and -//! the NF gates them on flow state. Port-forwarding sources stay out of the local -//! tables (a covering expose must answer for connection initiation), so a stage-2 -//! miss is reported distinctly (see [`LookupResult`]) for the NF to resolve -//! against flow state. +//! Masquerade destinations cannot accept new connections and port-forwarding +//! sources cannot initiate them, so neither may answer a lookup on its own. Their +//! rules are still in the tables, keyed on the revalidation information an +//! outdated flow supplies: the destination VPC for the former, the source NAT mode +//! for the latter. Each stage therefore asks what the flow vouches for first, and +//! runs a second, ungated pass over the misses. A packet on an established flow +//! keeps the peering and the NAT mode that flow was built on, even where an expose +//! covers the same address ungated; the ungated pass is what answers for forward +//! traffic, which carries no flow information of its own. +//! A stage-2 miss is still reported distinctly (see [`LookupResult`]): a packet +//! whose flow the NF, not the tables, has to resolve reaches it without any +//! revalidation information. use crate::{NatMode, NatRequirement}; use acl::dpdk::dyn_table::predicate_to_chunks; @@ -78,10 +83,12 @@ pub(crate) enum LookupResult { #[derive(Debug, Clone, Copy)] pub(crate) struct LookupInput { pub(crate) src_vpcd: VpcDiscriminant, + pub(crate) dst_vpcd: Option, pub(crate) src_ip: IpAddr, pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, + pub(crate) gate: SourceGate, } /// Result of a stage-1 (remote/destination) match. @@ -95,11 +102,13 @@ pub(super) struct Verdict { /// concrete; every other field is carried verbatim from the [`LookupInput`]). struct Query { src_vni: Vni, + dst_vni: GateVni, proto: NextHeader, src_ip: I, dst_ip: I, src_port: u16, dst_port: u16, + gate: SourceGate, } /// Lower a config L4 protocol to a bitmask predicate: a specific protocol matches exactly (every @@ -119,6 +128,60 @@ fn key_vni(vpcd: VpcDiscriminant) -> Vni { } } +/// The destination VPC a stage-1 rule is gated on, or `None` for a rule an ungated lookup reaches. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(transparent)] +pub(super) struct GateVni(pub(crate) Option); + +impl GateVni { + const UNGATED: Self = Self(None); + + fn is_gated(self) -> bool { + self.0.is_some() + } +} + +impl From> for GateVni { + fn from(vni: Option) -> Self { + Self(vni) + } +} + +impl FixedSize for GateVni { + const SIZE: usize = Vni::SIZE; + fn write_be(&self, out: &mut [u8]) { + self.0.map_or(0, Vni::as_u32).write_be(out); + } +} + +/// A flag to gate what a stage-2 entry answers for: +/// +/// - "revalidation of reply traffic associated with port-forwarding, with outdated flow info" +/// - everything else ([`Ungated`]) +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) enum SourceGate { + #[default] + Ungated, + PortFwdReply, +} + +impl SourceGate { + pub(crate) fn is_gated(self) -> bool { + self != Self::Ungated + } +} + +impl FixedSize for SourceGate { + const SIZE: usize = u8::SIZE; + fn write_be(&self, out: &mut [u8]) { + match self { + Self::Ungated => 0u8, + Self::PortFwdReply => 1u8, + } + .write_be(out); + } +} + // ------------------------------------------------------------------------------------------------- // Keys. // @@ -135,6 +198,9 @@ pub(super) struct RemoteKey { #[exact] #[cli(column_name = "src-vni")] src_vni: Vni, + #[exact] + #[cli(column_name = "dst-vni")] + dst_vni: GateVni, #[prefix] #[cli(column_name = "destination")] dst_ip: I, @@ -160,6 +226,9 @@ pub(super) struct LocalKey { #[range] #[cli(column_name = "src-port")] src_port: u16, + #[exact] + #[cli(column_name = "gate")] + gate: SourceGate, } // ------------------------------------------------------------------------------------------------- @@ -393,6 +462,7 @@ fn emit_remote( v4: &mut Vec, Verdict>>, v6: &mut Vec, Verdict>>, src_vni: Vni, + dst_vni: GateVni, ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, @@ -407,6 +477,7 @@ fn emit_remote( let rule = RemoteKeyRule:: { proto, src_vni: ExactSpec::new(src_vni), + dst_vni: ExactSpec::new(dst_vni), dst_ip: PrefixSpec::from(prefix), dst_port: port_range, }; @@ -421,6 +492,7 @@ fn emit_remote( let rule = RemoteKeyRule:: { proto, src_vni: ExactSpec::new(src_vni), + dst_vni: ExactSpec::new(dst_vni), dst_ip: PrefixSpec::from(prefix), dst_port: port_range, }; @@ -444,10 +516,12 @@ fn emit_local( ip_range: Prefix, port_range: RangeSpec, proto: MaskSpec, + gate: SourceGate, action: NatMode, ) { - // Port-forwarding sources are never emitted into the local tables, so the tie-break bit is - // always clear here; local rules keep pure prefix-length ordering. + // Port-forwarding sources are the only local rules a masquerade rule can overlap, and the + // "nat_mode" key already keeps the two apart, so the tie-break bit is always clear here; + // local rules keep pure prefix-length ordering. let priority = rule_priority(ip_range, false); match ip_range { Prefix::IPV4(prefix) => { @@ -457,6 +531,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, + gate: ExactSpec::new(gate), }; v4.push(NeutralRule { priority, @@ -472,6 +547,7 @@ fn emit_local( dst_vni: ExactSpec::new(dst_vni), src_ip: PrefixSpec::from(prefix), src_port: port_range, + gate: ExactSpec::new(gate), }; v6.push(NeutralRule { priority, @@ -497,7 +573,7 @@ impl RuleSet { for vpc in overlay.vpc_table().values() { let src_vni = vpc.vni(); for peering in vpc.peerings() { - let remote_vni = overlay.vpc_table().get_remote_vni(peering); + let remote_vni = peering.remote_vni(); let remote_vpcd = VpcDiscriminant::from_vni(remote_vni); let default_ip = || { if peering.is_v4() { @@ -508,21 +584,23 @@ impl RuleSet { }; // Stage 1: peer's public prefixes -> Verdict{dst VPC, dst NAT}. Masquerade - // destinations cannot receive connections, but their rules stay in the table: - // a masquerade Verdict lets the NF tell reply traffic on an established - // masquerade flow apart from a destination no peering covers (which must drop). - // The NF only accepts a masquerade Verdict when the packet rides such a flow. + // destinations cannot receive connections, so their rules are gated on the peer + // VNI: only a lookup revalidating against that very VPC -- reply traffic on an + // established masquerade flow -- reaches them. Everything else is ungated, which + // is what an ordinary lookup asks with. for expose in peering.remote().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = Verdict { nat_mode: NatRequirement::from_expose(expose), dst_vpcd: remote_vpcd, }; + let dst_vni = GateVni::from(expose.has_masquerade().then_some(remote_vni)); for prefix in expose.public_ips() { emit_remote( &mut rules.remote_v4, &mut rules.remote_v6, src_vni, + dst_vni, prefix.prefix(), prefix.into(), proto, @@ -535,6 +613,7 @@ impl RuleSet { &mut rules.remote_v4, &mut rules.remote_v6, src_vni, + GateVni::UNGATED, default_ip(), PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), @@ -546,15 +625,17 @@ 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()) - { + // cannot initiate connections, so, symmetrically, their rules are gated on the + // NAT mode they require: only a lookup revalidating against port forwarding + // reaches them. + for expose in peering.local().valexp() { let proto = proto_mask(expose.nat_proto().unwrap_or(L4Protocol::Any)); let action = NatRequirement::from_expose(expose); + let gate = if expose.has_port_forwarding() { + SourceGate::PortFwdReply + } else { + SourceGate::Ungated + }; for prefix in expose.ips() { emit_local( &mut rules.local_v4, @@ -564,6 +645,7 @@ impl RuleSet { prefix.prefix(), prefix.into(), proto, + gate, action, ); } @@ -577,6 +659,7 @@ impl RuleSet { default_ip(), PORT_RANGE_WILDCARD, proto_mask(L4Protocol::Any), + SourceGate::Ungated, None, ); } @@ -637,62 +720,130 @@ impl FlowFilterContext { // Single-key lookup: the readable per-packet oracle used by tests; production runs // lookup_batch. The differential test cross-checks the two against each other. #[cfg(test)] + #[allow(clippy::too_many_arguments)] pub(super) fn lookup( &self, src_vpcd: VpcDiscriminant, + dst_vpcd: Option, src_ip: IpAddr, dst_ip: IpAddr, proto: NextHeader, ports: Option<(u16, u16)>, + gate: SourceGate, ) -> LookupResult { let src_vni = key_vni(src_vpcd); + let dst_vni = GateVni::from(dst_vpcd.map(key_vni)); let (src_port, dst_port) = ports.unzip(); let src_port = src_port.unwrap_or(0); let dst_port = dst_port.unwrap_or(0); match (src_ip, dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { - let Some(verdict) = self.remote_v4.lookup(&RemoteKey { + let verdict = if let Some(v) = self.remote_v4.lookup(&RemoteKey { proto, src_vni, + dst_vni, dst_ip, dst_port, - }) else { - return LookupResult::DestinationMiss; + }) { + v + } else { + if dst_vni.is_gated() + && let Some(v) = self.remote_v4.lookup(&RemoteKey { + proto, + src_vni, + dst_vni: GateVni::UNGATED, + dst_ip, + dst_port, + }) + { + v + } else { + return LookupResult::DestinationMiss; + } }; + let dst_vni = key_vni(verdict.dst_vpcd); match self.local_v4.lookup(&LocalKey { proto, src_vni, - dst_vni: key_vni(verdict.dst_vpcd), + dst_vni, src_ip, src_port, + gate, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) + Some(src_nat_mode) => { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } + None => { + if gate.is_gated() + && let Some(src_nat_mode) = self.local_v4.lookup(&LocalKey { + proto, + src_vni, + dst_vni, + src_ip, + src_port, + gate: SourceGate::Ungated, + }) + { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } else { + LookupResult::SourceMiss(verdict.dst_vpcd) + } } - None => LookupResult::SourceMiss(verdict.dst_vpcd), } } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { - let Some(verdict) = self.remote_v6.lookup(&RemoteKey { + let verdict = if let Some(v) = self.remote_v6.lookup(&RemoteKey { proto, src_vni, + dst_vni, dst_ip, dst_port, - }) else { - return LookupResult::DestinationMiss; + }) { + v + } else { + if dst_vni.is_gated() + && let Some(v) = self.remote_v6.lookup(&RemoteKey { + proto, + src_vni, + dst_vni: GateVni::UNGATED, + dst_ip, + dst_port, + }) + { + v + } else { + return LookupResult::DestinationMiss; + } }; + let dst_vni = key_vni(verdict.dst_vpcd); match self.local_v6.lookup(&LocalKey { proto, src_vni, - dst_vni: key_vni(verdict.dst_vpcd), + dst_vni, src_ip, src_port, + gate, }) { - Some(nat_mode) => { - LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *nat_mode)) + Some(src_nat_mode) => { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } + None => { + if gate.is_gated() + && let Some(src_nat_mode) = self.local_v6.lookup(&LocalKey { + proto, + src_vni, + dst_vni, + src_ip, + src_port, + gate: SourceGate::Ungated, + }) + { + LookupResult::Route((verdict.dst_vpcd, verdict.nat_mode, *src_nat_mode)) + } else { + LookupResult::SourceMiss(verdict.dst_vpcd) + } } - None => LookupResult::SourceMiss(verdict.dst_vpcd), } } _ => { @@ -721,28 +872,34 @@ impl FlowFilterContext { out[i] = LookupResult::DestinationMiss; let proto = input.proto; let src_vni = key_vni(input.src_vpcd); + let dst_vni = GateVni::from(input.dst_vpcd.map(key_vni)); let (src_port, dst_port) = input.ports.unwrap_or((0, 0)); + let gate = input.gate; match (input.src_ip, input.dst_ip) { (IpAddr::V4(src_ip), IpAddr::V4(dst_ip)) => { v4_idx.push(i); v4_q.push(Query { src_vni, + dst_vni, proto, src_ip, dst_ip, src_port, dst_port, + gate, }); } (IpAddr::V6(src_ip), IpAddr::V6(dst_ip)) => { v6_idx.push(i); v6_q.push(Query { src_vni, + dst_vni, proto, src_ip, dst_ip, src_port, dst_port, + gate, }); } _ => { /* version mismatch: leave "out[i] = DestinationMiss" */ } @@ -764,7 +921,7 @@ fn lookup_versioned( idx: &[usize], out: &mut [LookupResult], ) where - RemoteKey: MatchKey, + RemoteKey: MatchKey + std::fmt::Debug, LocalKey: MatchKey, { for (q_chunk, i_chunk) in queries.chunks(MAX_BATCH).zip(idx.chunks(MAX_BATCH)) { @@ -774,6 +931,7 @@ fn lookup_versioned( .map(|q| RemoteKey { proto: q.proto, src_vni: q.src_vni, + dst_vni: q.dst_vni, dst_ip: q.dst_ip, dst_port: q.dst_port, }) @@ -781,7 +939,34 @@ fn lookup_versioned( let mut verdicts: Vec> = vec![None; q_chunk.len()]; remote.lookup_batch(&remote_keys, &mut verdicts); + // Reply traffic for masqueraded flows use the destination VNI as part of the key; this is + // to avoid conflicting entries if there are several VPCs exposing overlapping, masqueraded + // prefixes to a given VPC. If we have a destination VNI set here, we may be trying to + // re-validate a reply packet for a masqueraded flow (we're not sure of the direction, hence + // the first attempt with the destination VNI). Try again, without the destination VNI. + let mut reval_positions = Vec::new(); + let mut reval_keys = Vec::new(); + for (pos, (query, verdict)) in q_chunk.iter().zip(verdicts.iter_mut()).enumerate() { + if verdict.is_none() && query.dst_vni.is_gated() { + reval_positions.push(pos); + reval_keys.push(RemoteKey { + proto: query.proto, + src_vni: query.src_vni, + dst_vni: GateVni::UNGATED, + dst_ip: query.dst_ip, + dst_port: query.dst_port, + }); + } + } + let mut reval_verdicts = vec![None; reval_keys.len()]; + remote.lookup_batch(&reval_keys, &mut reval_verdicts); + for (pos, verdict) in reval_positions.into_iter().zip(reval_verdicts) { + verdicts[pos] = verdict; + } + // Stage 2: for the hits only, source -> source NAT. + // Port-forwarding rules use the NAT mode as part of the key, to dissociate keys from any + // keys associated to overlapping forward masquerade prefixes. let mut local_keys: Vec> = Vec::new(); let mut hit_pos: Vec = Vec::new(); for (pos, verdict) in verdicts.iter().enumerate() { @@ -793,6 +978,7 @@ fn lookup_versioned( dst_vni: key_vni(verdict.dst_vpcd), src_ip: q.src_ip, src_port: q.src_port, + gate: q.gate, }); hit_pos.push(pos); } @@ -800,6 +986,27 @@ fn lookup_versioned( let mut nat_modes: Vec> = vec![None; local_keys.len()]; local.lookup_batch(&local_keys, &mut nat_modes); + // Second pass: if nat_mode was set and we didn't find an entry for reply traffic associated + // with a port-forwarding flow, drop the gate to see if we have an entry for forward traffic + // for port-forwarding (forward traffic entries do not have flow-info nat mode attached, or + // we couldn't use it to initiate new flows). + let mut reval_positions = Vec::new(); + let mut reval_keys = Vec::new(); + for (pos, (nat_mode, &q_pos)) in nat_modes.iter().zip(hit_pos.iter()).enumerate() { + if nat_mode.is_none() && q_chunk[q_pos].gate.is_gated() { + reval_positions.push(pos); + reval_keys.push(LocalKey { + gate: SourceGate::Ungated, + ..local_keys[pos].clone() + }); + } + } + let mut reval_nat_modes = vec![None; reval_keys.len()]; + local.lookup_batch(&reval_keys, &mut reval_nat_modes); + for (pos, nat_mode) in reval_positions.into_iter().zip(reval_nat_modes) { + nat_modes[pos] = nat_mode; + } + // Scatter results back to the caller's output positions. A stage-1 miss stays // DestinationMiss; a stage-1 hit whose source matched nothing becomes SourceMiss. for (hit, &pos) in hit_pos.iter().enumerate() { @@ -832,9 +1039,9 @@ mod unit_tests { } #[test] - fn remote_key_has_four_fields_local_has_five() { - assert_eq!(RemoteKey::::N, 4); - assert_eq!(LocalKey::::N, 5); + fn remote_key_has_five_fields_local_has_six() { + assert_eq!(RemoteKey::::N, 5); + assert_eq!(LocalKey::::N, 6); } #[test] diff --git a/flow-filter/src/context/tests.rs b/flow-filter/src/context/tests.rs index c94dcd49bb..95d3495ad2 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -7,6 +7,7 @@ use super::LookupResult; use super::tables::RuleRow; +use super::tables::SourceGate; use crate::test_utils::*; use crate::{FlowFilterContext, NatMode, NatRequirement}; use lpm::prefix::L4Protocol; @@ -29,6 +30,25 @@ fn route( src_vpcd: VpcDiscriminant, headers: &Headers, ) -> Option { + match route_lookup(context, src_vpcd, None, SourceGate::Ungated, headers) { + LookupResult::Route((dst_vpcd, dst_nat, src_nat)) => Some(Route { + dst_vpcd, + dst_nat, + src_nat, + }), + LookupResult::SourceMiss(_) | LookupResult::DestinationMiss => None, + } +} + +// Extract the 5-tuple from headers (as the pipeline does) and run the route lookup for a packet +// originating from a given source VPC. +fn route_lookup( + context: &FlowFilterContext, + src_vpcd: VpcDiscriminant, + dst_vpcd: Option, + gate: SourceGate, + headers: &Headers, +) -> LookupResult { let net = headers.net().unwrap(); let src_ip = net.src_addr(); let dst_ip = net.dst_addr(); @@ -38,14 +58,7 @@ fn route( .map(NonZero::get) .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::SourceMiss(_) | LookupResult::DestinationMiss => None, - } + context.lookup(src_vpcd, dst_vpcd, src_ip, dst_ip, proto, ports, gate) } // ------------------------------------------------------------------------------------------------- @@ -186,8 +199,7 @@ fn overlapping_source_prefix_disambiguated_by_destination() { // We pin down which NAT requirement is returned for each end of a lookup. The source (local) end // carries private IPs; the destination (remote) end carries public IPs. Masquerade is only valid on // the source side (a masquerade destination cannot receive connections) and port forwarding only on -// the destination side (a port-forwarding source cannot initiate connections); these constraints -// are tested in `dst_side_nat_modes`. +// the destination side (a port-forwarding source cannot initiate connections). fn nat_modes_overlay() -> FlowFilterContext { context( @@ -285,16 +297,44 @@ fn dst_side_overlay() -> FlowFilterContext { fn dst_side_nat_modes() { let ctx = dst_side_overlay(); - // Masquerade destination: resolves at table level as a marker (the NF only lets it through - // for reply traffic on an established masquerade flow; see crate::tests) + // Masquerade source let masq = route( &ctx, - vpcd(100), - &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), + vpcd(200), + &build_tcp_packet(v4("192.168.70.1"), v4("10.0.0.5"), 1234, 5678), ) .expect("masquerade destination resolves as a marker"); - assert_eq!(masq.dst_vpcd, vpcd(200)); - assert_eq!(masq.dst_nat, Some(NatRequirement::Masquerade)); + assert_eq!(masq.dst_vpcd, vpcd(100)); + assert_eq!(masq.dst_nat, None); + assert_eq!(masq.src_nat, Some(NatRequirement::Masquerade)); + + // Masquerade destination: without the destination VPC discriminant hint, we fail to find the + // relevant destination entry. This is expected, because we can never initiate a flow in this + // direction, so we need the dst_vpcd from flow info to find the relevant entry (only necessary + // when re-validating after a configuration change). + let lookup_result = route_lookup( + &ctx, + vpcd(100), + None, + SourceGate::Ungated, + &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), + ); + assert_eq!(lookup_result, LookupResult::DestinationMiss); + + // Masquerade destination: With the destination VPC discriminant (virtually-)retrieved from flow + // information, we can determinate the right information for the packet. + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( + &ctx, + vpcd(100), + Some(vpcd(200)), + SourceGate::Ungated, + &build_tcp_packet(v4("10.0.0.5"), v4("70.0.0.10"), 1234, 5678), + ) else { + panic!("masquerade destination resolves as a marker"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(dst_nat, Some(NatRequirement::Masquerade)); + assert_eq!(src_nat, None); // Port-forwarding destination (matching proto + port): returned let pf = route( @@ -307,6 +347,30 @@ fn dst_side_nat_modes() { assert_eq!(pf.dst_nat, Some(NatRequirement::PortForwarding)); assert_eq!(pf.src_nat, None); + // Port-forwarding source without NAT mode hint: lookup fails + let lookup_result = route_lookup( + &ctx, + vpcd(200), + None, + SourceGate::Ungated, + &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), + ); + assert_eq!(lookup_result, LookupResult::SourceMiss(vpcd(100))); + + // Port-forwarding source without NAT mode hint: lookup fails + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( + &ctx, + vpcd(200), + None, + SourceGate::PortFwdReply, + &build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), + ) else { + panic!("masquerade destination resolves as a marker"); + }; + assert_eq!(dst_vpcd, vpcd(100)); + assert_eq!(dst_nat, None); + assert_eq!(src_nat, Some(NatRequirement::PortForwarding)); + // Port-forwarding destination, wrong port: no match. assert_eq!( route( @@ -428,12 +492,12 @@ fn port_forwarding_any_protocol_matches_tcp_and_udp() { } // ------------------------------------------------------------------------------------------------- -// Port forwarding is excluded from the source side (it cannot initiate connections). With both a -// masquerade and a port-forwarding expose on the source manifest, a source in the port-forwarding -// range is matched by masquerade instead. +// Port-forwarding and masquerade prefixes may overlap within a manifest. In this case, for +// re-validating return traffic for port-forwarding, we rely on the NAT mode from the flow +// information to make the distinction with similar-looking entries for forward masqueraded traffic. #[test] -fn source_port_forwarding_is_excluded_and_falls_back_to_masquerade() { +fn port_forwarding_and_masquerade_overlap_resoves_as_expected() { let ctx = context( &[("vpc1", 100), ("vpc2", 200)], vec![peering( @@ -455,15 +519,31 @@ fn source_port_forwarding_is_excluded_and_falls_back_to_masquerade() { )], ); // Source 1.0.0.27:2000 is inside the port-forwarding private range, yet resolves to masquerade. - let r = route( + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( &ctx, vpcd(100), + None, + SourceGate::PortFwdReply, &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), - ) - .expect("source resolves via the masquerade expose"); - assert_eq!(r.dst_vpcd, vpcd(200)); - assert_eq!(r.src_nat, Some(NatRequirement::Masquerade)); - assert_eq!(r.dst_nat, None); + ) else { + panic!("source resolves via the port-forwarding expose"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(src_nat, Some(NatRequirement::PortForwarding)); + assert_eq!(dst_nat, None); + + let LookupResult::Route((dst_vpcd, dst_nat, src_nat)) = route_lookup( + &ctx, + vpcd(100), + None, + SourceGate::Ungated, + &build_tcp_packet(v4("1.0.0.27"), v4("5.0.0.10"), 2000, 5678), + ) else { + panic!("source resolves via the masquerade expose"); + }; + assert_eq!(dst_vpcd, vpcd(200)); + assert_eq!(src_nat, Some(NatRequirement::Masquerade)); + assert_eq!(dst_nat, None); } // ------------------------------------------------------------------------------------------------- @@ -716,8 +796,24 @@ fn reference_and_dpdk_backends_agree() { for &(vni, src_ip, dst_ip, proto, ports) in probes { let src_vpcd = vpcd(vni); assert_eq!( - reference.lookup(src_vpcd, src_ip, dst_ip, proto, ports), - dpdk.lookup(src_vpcd, src_ip, dst_ip, proto, ports), + reference.lookup( + src_vpcd, + None, + src_ip, + dst_ip, + proto, + ports, + SourceGate::Ungated + ), + dpdk.lookup( + src_vpcd, + None, + src_ip, + dst_ip, + proto, + ports, + SourceGate::Ungated + ), "backends disagree on {src_ip} -> {dst_ip} ({proto:?}) from vni {vni}", ); } @@ -729,10 +825,12 @@ fn reference_and_dpdk_backends_agree() { .flatten() .map(|&(vni, src_ip, dst_ip, proto, ports)| LookupInput { src_vpcd: vpcd(vni), + dst_vpcd: None, src_ip, dst_ip, proto, ports, + gate: SourceGate::Ungated, }) .collect(); assert!(inputs.len() > 32, "want a multi-chunk batch"); @@ -746,10 +844,12 @@ fn reference_and_dpdk_backends_agree() { for (i, input) in inputs.iter().enumerate() { let single = reference.lookup( input.src_vpcd, + input.dst_vpcd, input.src_ip, input.dst_ip, input.proto, input.ports, + input.gate, ); assert_eq!(ref_out[i], single, "batched != single at index {i}"); } @@ -825,6 +925,7 @@ fn display_is_identical_across_backends() { "rank", "proto", "src-vni", + "dst-vni", "destination", "dst-port", "|", @@ -839,6 +940,7 @@ fn display_is_identical_across_backends() { "[0]", "TCP", "100", + "-", "80.0.0.5/32", "2222", "|", @@ -853,7 +955,7 @@ fn display_is_identical_across_backends() { assert_eq!( cells(&local_v4, "rank"), [ - "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "|", "NAT" + "rank", "proto", "src-vni", "dst-vni", "source", "src-port", "gate", "|", "NAT" ], "unexpected local heading row:\n{local_v4}" ); @@ -861,10 +963,10 @@ fn display_is_identical_across_backends() { // The heading assertions above would still pass if a section ran past its own table, since they // read its first heading row and stop. The ordering assertion below would not: it compares // positions, so a section that swallowed the table after it could order two rules that are not - // even in the same table and call the result precedence. Pin the boundary directly -- `dst-vni` + // even in the same table and call the result precedence. Pin the boundary directly -- `source` // is a local-table column, and the local table is the one that follows. assert!( - !remote_v4.contains("dst-vni"), + !remote_v4.contains("source"), "the remote v4 section ran past its own table:\n{remote_v4}" ); diff --git a/flow-filter/src/fuzz_gen.rs b/flow-filter/src/fuzz_gen.rs index 43b4bffdd5..097d2480df 100644 --- a/flow-filter/src/fuzz_gen.rs +++ b/flow-filter/src/fuzz_gen.rs @@ -23,6 +23,7 @@ #![cfg(test)] +use crate::context::SourceGate; use bolero::TypeGenerator; use config::external::overlay::vpc::{Vpc, VpcTable}; use config::external::overlay::vpcpeering::{VpcExpose, VpcManifest, VpcPeering, VpcPeeringTable}; @@ -135,6 +136,37 @@ impl ExposeSpec { ExposeSpec::PortForwarding(_) | ExposeSpec::PortFwProtoPair => None, } } + + /// Whether destinations of this expose are masquerade destinations: they cannot receive + /// connections, so the tables answer for them only when the lookup carries the destination VPC + /// an outdated flow revalidates against. + fn dest_needs_revalidation(self) -> bool { + matches!( + self, + ExposeSpec::Masquerade + | ExposeSpec::MasqueradeNestingPortFw(_) + | ExposeSpec::MasqueradeSameLenPortFw(_) + ) + } + + /// The port-forwarding exposes this spec contributes, as (protocol, host byte of an address + /// the private side covers). A port-forwarding source cannot initiate a connection, so the + /// tables answer for it only when the lookup carries the port-forwarding NAT mode an outdated + /// flow revalidates against -- and then in preference to any masquerade expose covering the + /// same address, which is why the specs that overlap the two are here too. + fn port_fw_sources(self) -> [Option<(FwProto, u8)>; 2] { + match self { + ExposeSpec::PortForwarding(proto) | ExposeSpec::MasqueradeNestingPortFw(proto) => { + [Some((proto, FW_HOST)), None] + } + // The port-forwarded prefix is the whole block, so any host of it will do. + ExposeSpec::MasqueradeSameLenPortFw(proto) => [Some((proto, 1)), None], + ExposeSpec::PortFwProtoPair => { + [Some((FwProto::Tcp, FW_HOST)), Some((FwProto::Udp, FW_HOST))] + } + ExposeSpec::Plain | ExposeSpec::StaticNat | ExposeSpec::Masquerade => [None, None], + } + } } /// An exclusion applied to an expose's private and public blocks. @@ -331,7 +363,7 @@ impl OverlaySpec { let remote = build_manifest(&vpc_name(b), &peering.remote, peering.v6, &mut blocks); derive_routing_probes( &mut routing_probes, - VNIS[a], + (VNIS[a], VNIS[b]), peering.v6, (local_base, &peering.local), (remote_base, &peering.remote), @@ -361,45 +393,98 @@ impl OverlaySpec { /// Append a routing probe for each compatible pair of local and remote exposes. /// -/// Port-forwarding probes target [`FW_HOST`] and [`FW_PUBLIC_PORTS`]. Other probes use host `.1`. +/// Port-forwarding destinations target [`FW_HOST`] and [`FW_PUBLIC_PORTS`], port-forwarding +/// sources the host their expose covers and [`FW_PRIVATE_PORTS`]. Other endpoints use host `.1`. +/// +/// Masquerade destinations and port-forwarding sources have no rule a plain lookup can reach: the +/// tables gate them on the revalidation information an outdated flow supplies, so their probes +/// carry it. Both stages ask the revalidated question first, so such a probe routes through the +/// gated rule even where a catch-all or a masquerade expose covers the same address. fn derive_routing_probes( out: &mut Vec, - src_vni: u32, + (src_vni, dst_vni): (u32, u32), v6: bool, (local_base, local): (u8, &ManifestSpec), (remote_base, remote): (u8, &ManifestSpec), ) { let src_vpcd = VpcDiscriminant::from_vni(Vni::new_checked(src_vni).unwrap()); + let dst_vpcd = VpcDiscriminant::from_vni(Vni::new_checked(dst_vni).unwrap()); + + // (address, port, protocol, revalidated source NAT mode) of every source the peering routes. + // A protocol of None leaves the choice to the destination. + let mut sources: Vec<(IpAddr, u16, Option, SourceGate)> = Vec::new(); for (li, lspec) in local.expose_specs().enumerate() { - if !lspec.source_capable() { - continue; + let block = local_base + li as u8; + if lspec.source_capable() { + sources.push(( + block_addr(block, 1, false, v6), + 1, + None, + SourceGate::Ungated, + )); } - let src_ip = block_addr(local_base + li as u8, 1, false, v6); - for (ri, rspec) in remote.expose_specs().enumerate() { - let dst_block = remote_base + ri as u8; - if let Some(dst_public) = rspec.dest_public_space() { - out.push(Probe { - src_vpcd, - src_ip, - dst_ip: block_addr(dst_block, 1, dst_public, v6), - proto: NextHeader::TCP, - ports: Some((1, 1)), - }); - } - for proto in rspec.portfw_protos() { - out.push(Probe { - src_vpcd, - src_ip, - dst_ip: block_addr(dst_block, FW_HOST, true, v6), - proto: proto.probe_next_header(), - // Source exposes do not constrain ports. - ports: Some((1, FW_PUBLIC_PORTS.0)), - }); - } + for (proto, host) in lspec.port_fw_sources().into_iter().flatten() { + sources.push(( + block_addr(block, host, false, v6), + FW_PRIVATE_PORTS.0, + Some(proto), + SourceGate::PortFwdReply, + )); + } + } + + // (address, port, protocol, revalidated destination VPC) of every destination the peering + // routes. + let mut destinations: Vec<(IpAddr, u16, Option, Option)> = Vec::new(); + for (ri, rspec) in remote.expose_specs().enumerate() { + let block = remote_base + ri as u8; + if let Some(dst_public) = rspec.dest_public_space() { + let revalidated = rspec.dest_needs_revalidation().then_some(dst_vpcd); + destinations.push((block_addr(block, 1, dst_public, v6), 1, None, revalidated)); + } + for proto in rspec.portfw_protos() { + destinations.push(( + block_addr(block, FW_HOST, true, v6), + FW_PUBLIC_PORTS.0, + Some(proto), + None, + )); + } + } + + for &(src_ip, src_port, src_proto, gate) in &sources { + for &(dst_ip, dst_port, dst_proto, revalidated_dst) in &destinations { + let Some(proto) = pair_proto(src_proto, dst_proto) else { + continue; + }; + out.push(Probe { + src_vpcd, + dst_vpcd: revalidated_dst, + src_ip, + dst_ip, + proto, + ports: Some((src_port, dst_port)), + gate, + }); } } } +/// The protocol a probe pairing these two endpoints carries, or `None` if each end constrains it +/// to a protocol the other rejects. An end that constrains nothing follows the other one, and a +/// pair that constrains nothing uses TCP. +fn pair_proto(src: Option, dst: Option) -> Option { + match (src, dst) { + (None, None) => Some(NextHeader::TCP), + (Some(proto), None) | (None, Some(proto)) => Some(proto.probe_next_header()), + (Some(FwProto::Any), Some(proto)) | (Some(proto), Some(FwProto::Any)) => { + Some(proto.probe_next_header()) + } + (Some(src), Some(dst)) if src == dst => Some(src.probe_next_header()), + (Some(_), Some(_)) => None, + } +} + fn vpc_name(index: usize) -> String { format!("vpc{}", index + 1) } @@ -608,28 +693,46 @@ pub(crate) struct ProbeSpec { proto: ProbeProto, sport: PortSel, dport: PortSel, + /// Revalidation information, as an outdated flow would supply it: the destination VPC the + /// flow was routed to, and whether it is a port-forwarding flow. Both are drawn freely (any + /// VPC, either gate, neither), so the lookups see combinations no flow would produce as well. + revalidate_dst: Option, + revalidate_port_fw: bool, } /// A resolved probe: the arguments of one route lookup. #[derive(Debug, Clone, Copy)] pub(crate) struct Probe { pub(crate) src_vpcd: VpcDiscriminant, + pub(crate) dst_vpcd: Option, pub(crate) src_ip: IpAddr, pub(crate) dst_ip: IpAddr, pub(crate) proto: NextHeader, pub(crate) ports: Option<(u16, u16)>, + pub(crate) gate: SourceGate, +} + +/// The VPC discriminant a `u8` selector draws: one of the generated VPCs, or the bogus one. +fn vpcd_from_sel(sel: u8) -> VpcDiscriminant { + let vni = match sel as usize % (VNIS.len() + 1) { + i if i < VNIS.len() => VNIS[i], + _ => BOGUS_VNI, + }; + VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap()) } impl ProbeSpec { pub(crate) fn resolve(&self, blocks: u8) -> Probe { let nblocks = blocks.max(1); - let vni = match self.vni_sel as usize % (VNIS.len() + 1) { - i if i < VNIS.len() => VNIS[i], - _ => BOGUS_VNI, - }; let dst_v6 = self.v6 ^ self.cross_version; Probe { - src_vpcd: VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap()), + src_vpcd: vpcd_from_sel(self.vni_sel), + dst_vpcd: self.revalidate_dst.map(vpcd_from_sel), + gate: if self.revalidate_port_fw { + SourceGate::PortFwdReply + } else { + SourceGate::Ungated + }, src_ip: block_addr( self.src_block % nblocks, self.src_host, diff --git a/flow-filter/src/lib.rs b/flow-filter/src/lib.rs index 7aa534d111..1674eb8309 100644 --- a/flow-filter/src/lib.rs +++ b/flow-filter/src/lib.rs @@ -29,7 +29,7 @@ pub use context::{ FlowFilterContext, FlowFilterContextReader, FlowFilterContextReaderFactory, FlowFilterContextWriter, }; -use context::{LookupInput, LookupResult}; +use context::{LookupInput, LookupResult, SourceGate}; pub struct FlowFilter { name: String, @@ -118,6 +118,7 @@ impl FlowFilter { genid: i64, ) -> Classification { let nfi = &self.name; + let (mut revalidation_dst_vpcd, mut revalidation_gate) = (None, SourceGate::Ungated); let attached_flow = FlowSummary::from_meta(packet.meta()); if let Some(flow_summary) = attached_flow.as_ref() { // Bypass flow-filter if packet has up-to-date active flow-info @@ -125,6 +126,8 @@ impl FlowFilter { Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow_summary); return Classification::Bypassed; } + (revalidation_dst_vpcd, revalidation_gate) = + self.flow_revalidation_data(flow_summary, genid); } let Some(net) = packet.try_ip() else { @@ -140,6 +143,7 @@ impl FlowFilter { let input = LookupInput { src_vpcd, + dst_vpcd: revalidation_dst_vpcd, src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -148,6 +152,7 @@ impl FlowFilter { .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }), + gate: revalidation_gate, }; Classification::Lookup { input, @@ -156,14 +161,6 @@ impl FlowFilter { } /// Phase C: apply a resolved route (or drop on a miss) to a single packet. - /// - /// The tables cannot answer for reply traffic of established stateful-NAT sessions: masquerade - /// destinations only appear as marker rules (they cannot accept new connections) and - /// port-forwarding sources are absent altogether (they cannot initiate). For those two cases - /// -- and only those -- an active flow carrying the matching NAT state lets the packet - /// through, exactly as the flow-bypass path would. The flow's validity under the new - /// configuration remains the stateful NFs' responsibility; a genuine miss (no peering covers - /// the packet) still drops and invalidates. fn apply_route( &self, packet: &mut Packet, @@ -175,15 +172,6 @@ impl FlowFilter { let (dst_vpcd, dst_nat_mode, src_nat_mode) = match result { LookupResult::Route(route) => route, LookupResult::SourceMiss(dst_vpcd) => { - // Port-forwarding sources are deliberately absent from the local tables; reply - // traffic from one rides its established flow. - if let Some(flow) = - active_stateful_flow(flow_summary, dst_vpcd, |f| f.needs_port_forwarding) - { - debug!("{nfi}: Source allowed by established port-forwarding flow"); - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow); - return; - } debug!("{nfi}: Source not allowed towards {dst_vpcd}, dropping packet"); packet.invalidate_flows(); packet.done(DoneReason::Filtered); @@ -196,25 +184,6 @@ impl FlowFilter { return; } }; - - // A masquerade destination cannot accept new connections; its rule is in the table only - // so that reply traffic on an established masquerade flow is distinguishable from a - // destination no peering covers. - if dst_nat_mode == Some(NatRequirement::Masquerade) { - if let Some(flow) = active_stateful_flow(flow_summary, dst_vpcd, |f| f.needs_masquerade) - { - debug!("{nfi}: Masquerade destination allowed by established flow"); - Self::tag_for_bypass(packet.meta_mut(), dst_vpcd, flow); - return; - } - debug!( - "{nfi}: Masquerade destination with no established flow, dropping packet (cannot initiate a connection towards a masquerade expose)" - ); - packet.invalidate_flows(); - packet.done(DoneReason::Filtered); - return; - } - debug!( "{nfi}: Packet matches peering configuration, found VPC {dst_vpcd} and NAT modes {src_nat_mode:?} (src), {dst_nat_mode:?} (dst)" ); @@ -300,7 +269,7 @@ impl FlowFilter { return false; } let (nfi, flowkey) = (&self.name, flow_summary.flow_info.flowkey()); - if flow_summary.dst_vpcd != Some(new_dst_vpcd) { + if flow_summary.dst_vpcd != new_dst_vpcd { debug!("{nfi}: Outdated flow {flowkey} (new dst: {new_dst_vpcd}) will be invalidated."); return true; } @@ -324,6 +293,33 @@ impl FlowFilter { false } + fn flow_revalidation_data( + &self, + flow_summary: &FlowSummary, + genid: i64, + ) -> (Option, SourceGate) { + // The only case when we need re-validation is when we have an active flow with an outdated + // genid. If this is not the case, return None. + if flow_summary.flow_info.status() != FlowStatus::Active + || flow_summary.flow_info.genid() >= genid + { + return (None, SourceGate::Ungated); + } + if flow_summary.needs_masquerade { + // If we need revalidation and the flow is masqueraded, we may need the destination VPC + // id from the flow to look up for reverse traffic's entry in the "remote" table. + (Some(flow_summary.dst_vpcd), SourceGate::Ungated) + } else if flow_summary.needs_port_forwarding { + // We need revalidation and the flow is with port-forwarding, although we don't know if + // it's on the source (reply traffic) or destination side (forward traffic). In doubt, + // turn on the gate to enable looking up for entries for reply port-forwarding traffic + // in the "local"-side context table. + (None, SourceGate::PortFwdReply) + } else { + (None, SourceGate::Ungated) + } + } + fn dst_vpcd_from_valid_flow( &self, flow_summary: &FlowSummary, @@ -342,18 +338,10 @@ impl FlowFilter { return None; } - let Some(dst_vpcd) = flow_summary.dst_vpcd else { - debug!( - "{nfi}: Flow information does not specify destination VPC. This is a bug. Ignoring it..." - ); - flow_summary.flow_info.invalidate_pair(); - return None; - }; - // Current and newer-generation flows bypass the filter. Workers may observe a new config // generation after flows have already been stamped with it. debug!("{nfi}: Packet can bypass flow filter thanks to flow information"); - Some(dst_vpcd) + Some(flow_summary.dst_vpcd) } } @@ -378,7 +366,7 @@ impl NetworkFunction for FlowFilter { #[derive(Debug, Clone)] struct FlowSummary { genid: i64, - dst_vpcd: Option, + dst_vpcd: VpcDiscriminant, needs_masquerade: bool, needs_port_forwarding: bool, flow_info: Arc, @@ -386,13 +374,16 @@ struct FlowSummary { impl FlowSummary { fn from_meta(meta: &PacketMeta) -> Option { - let Some(flow_info) = &meta.flow_info else { + let flow_info = meta.flow_info.as_ref()?; + let locked_info = flow_info.locked.read(); + let Some(dst_vpcd) = locked_info.dst_vpcd else { + debug!("Flow info lacks destination VPC. This is a bug. Invalidating flow.."); + flow_info.invalidate_pair(); return None; }; - let locked_info = flow_info.locked.read(); Some(Self { genid: flow_info.genid(), - dst_vpcd: locked_info.dst_vpcd, + dst_vpcd, needs_masquerade: locked_info.nat_state.is_some(), needs_port_forwarding: locked_info.port_fw_state.is_some(), flow_info: flow_info.clone(), @@ -400,23 +391,6 @@ impl FlowSummary { } } -/// The flow, if it is active, agrees with the lookup on the destination VPC, and carries the -/// stateful-NAT state selected by `has_state`. Such a flow vouches for reply traffic that the -/// tables cannot answer for (see [`FlowFilter::apply_route`]). No genid check: an up-to-date flow -/// would have bypassed the lookup already, and an outdated one is exactly the case where the flow -/// must speak for the packet; the stateful NFs remain the authority on the state itself. -fn active_stateful_flow( - flow_summary: Option<&FlowSummary>, - dst_vpcd: VpcDiscriminant, - has_state: impl Fn(&FlowSummary) -> bool, -) -> Option<&FlowSummary> { - flow_summary.filter(|flow| { - flow.flow_info.status() == FlowStatus::Active - && flow.dst_vpcd == Some(dst_vpcd) - && has_state(flow) - }) -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum NatRequirement { Static, diff --git a/flow-filter/src/tests.rs b/flow-filter/src/tests.rs index cb9222387b..091f7c1627 100644 --- a/flow-filter/src/tests.rs +++ b/flow-filter/src/tests.rs @@ -5,6 +5,7 @@ #![cfg(test)] +use crate::context::SourceGate; use crate::context::{FlowFilterContext, FlowFilterContextWriter}; use crate::fuzz_gen::Probe; use crate::test_utils::{ @@ -566,11 +567,7 @@ fn outdated_flow_with_consistent_state_is_kept() { } // ------------------------------------------------------------------------------------------------- -// Stateful reply traffic across config changes. The tables cannot answer for the reverse direction -// of stateful-NAT sessions (masquerade destinations are only markers, port-forwarding sources are -// absent altogether), so after a genid bump those packets must ride their established flow instead -// of being dropped -- while packets with no such flow, and flows whose peering is gone, still fail -// closed. +// Stateful reply traffic across config changes. #[test] fn masquerade_reply_on_established_flow_survives_config_change() { @@ -626,7 +623,6 @@ fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { Some(vpcd(200)), build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), ); - // The flow's recorded destination does not match what the tables resolve: stale, drop. let flow = attach_flow(&mut p, Some(vpcd(300)), true, true, false); let out = run(&mut flow_filter, p); assert_eq!(out.get_done(), Some(DoneReason::Filtered)); @@ -637,8 +633,7 @@ fn masquerade_reply_with_mismatched_flow_destination_is_filtered() { fn port_forwarding_reply_on_established_flow_survives_config_change() { let (mut flow_filter, _) = make_flow_filter(dst_port_forwarding_context()); set_genid(&mut flow_filter, 5); - // Reply direction of a forwarded session: the forwarded host answers from its private - // address, which is (deliberately) not in the local tables. + // Reply direction of a forwarded session: the forwarded host answers from its private address. let mut p = packet( Some(vpcd(200)), build_tcp_packet(v4("192.168.80.5"), v4("10.0.0.5"), 22, 1234), @@ -681,6 +676,282 @@ fn stateful_flow_does_not_survive_peering_removal() { assert_eq!(flow.status(), FlowStatus::Cancelled); } +// ------------------------------------------------------------------------------------------------- +// Config update and flow re-validation in the case of overlaps + +#[test] +fn revalidation_works_in_case_of_remote_masquerade_overlap() { + let ctx = context( + &[("vpc1", 100), ("vpc2", 200), ("vpc3", 300)], + // vpc2 and vpc3 both expose the same masqueraded prefixes towards the same vpc1 prefix + vec![ + peering( + "vpc1-to-vpc2", + ("vpc1", vec![expose("1.0.0.0/24")]), + ("vpc2", vec![expose_masquerade("2.0.0.0/24", "10.0.0.0/24")]), + ), + peering( + "vpc1-to-vpc3", + ("vpc1", vec![expose("1.0.0.0/24")]), + ("vpc3", vec![expose_masquerade("2.0.0.0/24", "10.0.0.0/24")]), + ), + ], + ); + let (mut flow_filter, writer) = make_flow_filter(ctx); + + // Initial packet from vpc2 to vpc1 (no flow info) passes + let p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + + // Reply from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Request from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Bump flow-filter genid + set_genid(&mut flow_filter, 5); + + // Reply from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Reply from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("1.0.0.1"), 2222, 1111), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Rmove peering, bump genid again + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, 6); + + // Reply from vpc1 to vpc2 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("10.0.0.1"), 1111, 2222), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + +#[test] +fn revalidation_works_in_case_of_local_masquerade_portforwarding_overlap() { + let ctx = context( + &[("vpc1", 100), ("vpc2", 200)], + // vpc1 uses overlapping prefixes for masquerade and port-forwarding + vec![peering( + "vpc1-to-vpc2", + ( + "vpc1", + vec![ + expose_port_forwarding( + "1.0.0.0/24", + (2000, 3000), + "10.0.0.0/24", + (5000, 6000), + Some(L4Protocol::Tcp), + ), + expose_masquerade("1.0.0.0/25", "10.0.0.0/25"), + ], + ), + ("vpc2", vec![expose("2.0.0.0/24")]), + )], + ); + let (mut flow_filter, writer) = make_flow_filter(ctx); + + // Port-forwarding: Initial packet from vpc2 to vpc1 (no flow info) passes + let p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + + // Port-forwarding: Reply from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Port-forwarding: Request from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Masquerade: Initial packet from vpc1 to vpc2 (no flow info) passes + let p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + + // Masquerade: Reply from vpc2 to vpc1 (with flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Request from vpc1 to vpc2 (with flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Bump flow-filter genid + set_genid(&mut flow_filter, 5); + + // ------ + + // Port-forwarding: Reply from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Port-forwarding: Request from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, false, true); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_port_forwarding()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // ------ + + // Masquerade: Reply from vpc2 to vpc1 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(100))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Request from vpc1 to vpc2 (with outdated flow info) passes + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, true, false); + let out = run(&mut flow_filter, p); + assert!(!out.is_done(), "{:?}", out.get_done()); + assert_eq!(out.meta().dst_vpcd, Some(vpcd(200))); + assert!(out.meta().requires_masquerade()); + assert_ne!(flow.status(), FlowStatus::Cancelled); + + // Rmove peering, bump genid again + writer.store(context(&[], vec![])); + set_genid(&mut flow_filter, 6); + + // Port-forwarding: Reply from vpc1 to vpc2 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(100)), + build_tcp_packet(v4("1.0.0.1"), v4("2.0.0.1"), 2000, 8000), + ); + let flow = attach_flow(&mut p, Some(vpcd(200)), true, false, true); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); + + // Masquerade: Reply from vpc2 to vpc1 (with outdated flow info) is dropped, flow cancelled + let mut p = packet( + Some(vpcd(200)), + build_tcp_packet(v4("2.0.0.1"), v4("10.0.0.1"), 8000, 5000), + ); + let flow = attach_flow(&mut p, Some(vpcd(100)), true, true, false); + let out = run(&mut flow_filter, p); + assert_eq!(out.get_done(), Some(DoneReason::Filtered)); + assert_eq!(flow.status(), FlowStatus::Cancelled); +} + // ------------------------------------------------------------------------------------------------- // Stateful flows: flow-key attachment for the {masquerade|port-forwarding} + static-NAT combination @@ -850,23 +1121,24 @@ struct InvalidationCase { meta_port_forwarding: bool, has_flow: bool, genid: GenidRel, - /// `Some(true)`: the flow's destination equals the route's; `Some(false)`: a different one; - /// `None`: the flow records no destination. - flow_dst_matches: Option, + /// `true`: the flow's destination equals the route's; `false`: a different one. A flow that + /// records no destination at all cannot reach this decision: `FlowSummary::from_meta` + /// invalidates it and reports no summary, which the `has_flow: false` cases already cover. + flow_dst_matches: bool, flow_masquerade: bool, flow_port_forwarding: bool, } // The specification: a flow is invalidated iff it comes from a DIFFERENT config generation // (older or newer -- only an equal genid is trusted) AND the filter can prove it stale: the -// destination changed (or was never recorded), a stateful-NAT requirement appeared or -// disappeared, or the route no longer needs state at all. Anything else is deferred to the -// stateful NFs, which own the state's validity. +// destination changed, a stateful-NAT requirement appeared or disappeared, or the route no longer +// needs state at all. Anything else is deferred to the stateful NFs, which own the state's +// validity. fn expected_invalidation(case: &InvalidationCase) -> bool { if !case.has_flow || matches!(case.genid, GenidRel::Same) { return false; } - case.flow_dst_matches != Some(true) + !case.flow_dst_matches || case.meta_masquerade != case.flow_masquerade || case.meta_port_forwarding != case.flow_port_forwarding || (!case.meta_masquerade && !case.meta_port_forwarding) @@ -903,10 +1175,10 @@ fn invalidation_decision_matches_spec() { GenidRel::Same => GENID, GenidRel::Newer => GENID + 3, }, - dst_vpcd: match case.flow_dst_matches { - Some(true) => Some(route_dst), - Some(false) => Some(vpcd(300)), - None => None, + dst_vpcd: if case.flow_dst_matches { + route_dst + } else { + vpcd(300) }, needs_masquerade: case.flow_masquerade, needs_port_forwarding: case.flow_port_forwarding, @@ -1118,6 +1390,9 @@ fn probe_from_packet(pkt: &Packet, src_vpcd: VpcDiscriminant) -> Opt let net = pkt.try_ip()?; Some(Probe { src_vpcd, + // These packets belong to no flow, so they don't need flow revalidation info. + dst_vpcd: None, + gate: SourceGate::Ungated, src_ip: net.src_addr(), dst_ip: net.dst_addr(), proto: net.next_header(), @@ -1151,6 +1426,11 @@ fn probe_packet(probe: &Probe) -> Option<(Packet, Probe)> { use net::ip::NextHeader; let mut probe = *probe; + // The NF sees these packets without a flow, so the revalidation information a derived probe + // carries never reaches the lookup: the oracle must not try to lookup for revalidation info + // that the NF cannot see, so we clear revalidation info. + probe.dst_vpcd = None; + probe.gate = SourceGate::Ungated; if let Some((sport, dport)) = probe.ports.as_mut() { *sport = (*sport).max(1); *dport = (*dport).max(1);