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 8d790713ae..a842668468 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}; @@ -20,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; // ------------------------------------------------------------------------------------------------- @@ -50,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") @@ -81,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; } @@ -107,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; } @@ -128,17 +140,17 @@ 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((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/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 d8e98fabf5..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; @@ -56,12 +62,32 @@ 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); -/// 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). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub 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 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. @@ -387,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( @@ -446,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:: { @@ -507,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), @@ -545,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() { @@ -667,9 +707,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 +731,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 +850,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), }; @@ -894,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 c94dcd49bb..f763c96120 100644 --- a/flow-filter/src/context/tests.rs +++ b/flow-filter/src/context/tests.rs @@ -7,28 +7,21 @@ 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 +// 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(); @@ -38,12 +31,18 @@ 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, - }), + 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, } } @@ -98,8 +97,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 +136,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 +228,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 @@ -285,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, 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( @@ -304,8 +303,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!( @@ -318,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. @@ -334,7 +378,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 +389,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 +435,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 +467,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 +506,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 +596,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 +606,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() { @@ -868,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 c2250bde67..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; +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 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 - 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); - 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,78 +171,159 @@ impl FlowFilter { .map(NonZero::get) .zip(t.dst_port().map(NonZero::get)) }), - }; - Classification::Lookup { - input, - flow_summary: attached_flow, } } - /// 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 (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; + + 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, } - }; + } + } - // 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 @@ -236,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); } @@ -290,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; } @@ -323,38 +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 { - 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 { - debug!( - "{nfi}: Flow information does not specify destination 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 - debug!("{nfi}: Packet can bypass flow filter thanks to flow information"); - Some(dst_vpcd) - } } impl NetworkFunction for FlowFilter { @@ -377,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(), @@ -400,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 2e2c84107b..a39b5b0c0d 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; @@ -22,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 @@ -37,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) { @@ -181,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( @@ -199,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( @@ -213,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( @@ -286,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( @@ -302,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()); @@ -332,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 @@ -350,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. @@ -448,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. @@ -464,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 @@ -513,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 @@ -524,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()); @@ -531,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. @@ -541,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. @@ -565,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))); @@ -590,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); @@ -612,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); @@ -625,17 +860,22 @@ 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); + flow.set_genid(5); 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] +#[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( @@ -643,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))); @@ -663,19 +904,372 @@ fn port_forwarding_reply_without_flow_is_filtered() { assert_eq!(out.get_done(), Some(DoneReason::Filtered)); } +// (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 stateful_flow_does_not_survive_peering_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()); +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); +} + +// ------------------------------------------------------------------------------------------------- +// 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); - let mut p = packet( + + // 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)), - build_tcp_packet(v4("5.0.0.10"), v4("30.0.0.5"), 5678, 1234), + 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); } @@ -836,36 +1430,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) @@ -897,23 +1484,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:?}", ); @@ -1040,10 +1628,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() @@ -1052,7 +1650,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] { 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();