diff --git a/Cargo.lock b/Cargo.lock index 23d39ecd15..01fa305f6a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1701,6 +1701,7 @@ dependencies = [ "dataplane-config", "dataplane-flow-entry", "dataplane-flow-filter", + "dataplane-k8s-intf", "dataplane-lpm", "dataplane-net", "dataplane-pipeline", diff --git a/k8s-intf/src/bolero/crd.rs b/k8s-intf/src/bolero/crd.rs index 781a0407b0..00e95b3a16 100644 --- a/k8s-intf/src/bolero/crd.rs +++ b/k8s-intf/src/bolero/crd.rs @@ -128,6 +128,9 @@ mod test { // empty list satisfies that assertion vacuously, so require a non-empty one to be // reachable here, where the reachability claims live. "bgp neighbors", + // `nat`'s allocator-migration concurrency test needs a config whose peerings actually + // masquerade; without one the allocator is never installed and the test is vacuous. + "masquerade expose", ]; // An atomic bitmask rather than a collection behind a lock: `bolero` runs each case inside @@ -171,6 +174,12 @@ mod test { if exposes.len() > 1 { record("multiple exposes"); } + if exposes + .iter() + .any(|e| e.nat.as_ref().is_some_and(|n| n.masquerade.is_some())) + { + record("masquerade expose"); + } } } } diff --git a/k8s-intf/src/bolero/expose.rs b/k8s-intf/src/bolero/expose.rs index 180094eb55..6e799d3904 100644 --- a/k8s-intf/src/bolero/expose.rs +++ b/k8s-intf/src/bolero/expose.rs @@ -53,6 +53,7 @@ pub fn default_expose() -> GatewayAgentPeeringsPeeringExpose { pub struct LegalValueExposeGenerator<'a> { subnets: &'a SubnetMap, pool: Option<&'a PrefixPool>, + allow_masquerade: bool, } impl<'a> LegalValueExposeGenerator<'a> { @@ -61,9 +62,20 @@ impl<'a> LegalValueExposeGenerator<'a> { Self { subnets, pool: None, + allow_masquerade: false, } } + /// Permit `from_pool` mode to generate masquerade exposes. + /// + /// `Peering::check_nat_modes` forbids masquerade on *both* sides of a peering, so this is not + /// decidable from one manifest and must be enabled by the caller that owns both sides. + #[must_use] + pub fn allow_masquerade(mut self) -> Self { + self.allow_masquerade = true; + self + } + /// Draw addresses from `pool`, and restrict generation to exposes that pass validation. /// /// `VpcExpose::validate` couples an expose's fields together in ways that cannot be satisfied @@ -76,24 +88,29 @@ impl<'a> LegalValueExposeGenerator<'a> { /// - masquerade forbids port ranges entirely; /// - and, across the whole manifest, no two exposes may overlap. /// - /// This mode satisfies all of them by generating the tractable subset: one address family per - /// expose, `ips` drawn from the shared pool (so overlap is impossible by construction), and no - /// exclusions, NAT, or `vpcSubnet` references. + /// This mode satisfies them by generating a tractable subset: one address family per expose, + /// addresses drawn from the shared pool (so overlap is impossible by construction), and no + /// exclusions or `vpcSubnet` references. Of the NAT modes only masquerade is generated, since + /// its extra obligations -- an `as` range in the same family, and no port ranges anywhere -- are + /// the only ones satisfiable without also controlling prefix sizes. /// - /// FIXME: extending this to NAT means generating `ips`/`as` in matched sizes per NAT mode, and - /// to `vpcSubnet` references means coordinating the VPC subnet generator with the same pool. - /// Until then, those shapes are exercised at the conversion level only - /// (`converters::k8s::config::expose`). + /// FIXME: static NAT additionally requires `ips` and `as` to cover the *same number of + /// addresses*, and port forwarding requires exactly one prefix per side with matching port + /// ranges; both need the pool to hand out sized blocks on request. `vpcSubnet` references need + /// the VPC subnet generator to draw from this same pool. Until then those shapes are exercised + /// at the conversion level only (`converters::k8s::config::expose`). #[must_use] pub fn from_pool(mut self, pool: &'a PrefixPool) -> Self { self.pool = Some(pool); self } - /// The validation-legal subset: a single family, pool-allocated `ips`, nothing else. + /// The validation-legal subset: a single family and pool-allocated addresses, with either no + /// NAT or masquerade. fn generate_from_pool( d: &mut D, pool: &PrefixPool, + allow_masquerade: bool, ) -> Option { let v4 = d.gen_bool(None)?; let count = d.gen_u16(Bound::Included(&1), Bound::Included(&4))?; @@ -106,12 +123,44 @@ impl<'a> LegalValueExposeGenerator<'a> { vpc_subnet: None, }) .collect(); + + // Masquerade needs a non-empty `as` range, in the same family as `ips` and with no port + // ranges. Its size need not match `ips` -- that is a static-NAT obligation -- so the two + // sides are drawn independently from the pool. + let masquerade = allow_masquerade && d.gen_bool(None)?; + let (r#as, nat) = if masquerade { + let as_count = d.gen_u16(Bound::Included(&1), Bound::Included(&4))?; + let as_prefixes = pool + .take(as_count, v4) + .into_iter() + .map(|cidr| GatewayAgentPeeringsPeeringExposeAs { + cidr: Some(cidr), + not: None, + }) + .collect(); + let idle_timeout_secs = d.gen_u64(Bound::Included(&1), Bound::Included(&(2 * 3600)))?; + ( + Some(as_prefixes), + Some(GatewayAgentPeeringsPeeringExposeNat { + masquerade: Some(GatewayAgentPeeringsPeeringExposeNatMasquerade { + idle_timeout: Some( + std::time::Duration::from_secs(idle_timeout_secs).into(), + ), + }), + port_forward: None, + r#static: None, + }), + ) + } else { + (None, None) + }; + Some(GatewayAgentPeeringsPeeringExpose { - r#as: None, + r#as, ips: Some(ips), // Explicit `false` is legal and takes the same path as absent; cover both spellings. default: d.gen_bool(None)?.then_some(false), - nat: None, + nat, }) } } @@ -121,7 +170,7 @@ impl ValueGenerator for LegalValueExposeGenerator<'_> { fn generate(&self, d: &mut D) -> Option { if let Some(pool) = self.pool { - return Self::generate_from_pool(d, pool); + return Self::generate_from_pool(d, pool, self.allow_masquerade); } let num_ips = d.gen_u16(Bound::Included(&1), Bound::Included(&16))?; diff --git a/k8s-intf/src/bolero/peering.rs b/k8s-intf/src/bolero/peering.rs index 60eb20ec30..4fe280aca1 100644 --- a/k8s-intf/src/bolero/peering.rs +++ b/k8s-intf/src/bolero/peering.rs @@ -25,6 +25,7 @@ use crate::gateway_agent_crd::{GatewayAgentPeerings, GatewayAgentPeeringsPeering pub struct LegalValuePeeringsPeeringGenerator<'a> { subnets: &'a SubnetMap, allow_default: bool, + allow_masquerade: bool, pool: Option<&'a PrefixPool>, } @@ -34,6 +35,7 @@ impl<'a> LegalValuePeeringsPeeringGenerator<'a> { Self { subnets, allow_default: false, + allow_masquerade: false, pool: None, } } @@ -54,6 +56,16 @@ impl<'a> LegalValuePeeringsPeeringGenerator<'a> { self.pool = Some(pool); self } + + /// Permit this side to masquerade. + /// + /// The caller must not enable this for both sides of the same peering; see + /// [`crate::bolero::expose::LegalValueExposeGenerator::allow_masquerade`]. + #[must_use] + pub fn allow_masquerade(mut self) -> Self { + self.allow_masquerade = true; + self + } } impl ValueGenerator for LegalValuePeeringsPeeringGenerator<'_> { @@ -66,6 +78,11 @@ impl ValueGenerator for LegalValuePeeringsPeeringGenerator<'_> { Some(pool) => expose_gen.from_pool(pool), None => expose_gen, }; + let expose_gen = if self.allow_masquerade { + expose_gen.allow_masquerade() + } else { + expose_gen + }; let mut expose = (0..num_expose) .map(|_| expose_gen.generate(d)) .collect::>>()?; @@ -191,6 +208,16 @@ impl LegalValuePeeringsGenerator<'_> { _ => (first, second), }; + // `Peering::check_nat_modes` forbids masquerade on both sides at once, so at most one side + // may masquerade. `2` means neither. Independent of `default_side`: a manifest may hold + // both a default expose and a masquerading one. + let masquerade_side = d.gen_usize(Bound::Included(&0), Bound::Included(&2))?; + let (first, second) = match masquerade_side { + 0 => (first.allow_masquerade(), second), + 1 => (first, second.allow_masquerade()), + _ => (first, second), + }; + let peering = BTreeMap::from([ (vpc_names[0].to_string(), first.generate(d)?), (vpc_names[1].to_string(), second.generate(d)?), diff --git a/nat/Cargo.toml b/nat/Cargo.toml index 3c7378da7b..1b27995185 100644 --- a/nat/Cargo.toml +++ b/nat/Cargo.toml @@ -41,8 +41,9 @@ fixin = { workspace = true } flow-filter = { workspace = true } test-utils = { workspace = true } lpm = { workspace = true, features = ["testing"] } +k8s-intf = { workspace = true, features = ["bolero"] } net = { workspace = true, features = ["bolero"] } -tokio = { workspace = true, features = ["macros", "rt", "time"] } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time"] } tracectl = { workspace = true } # external diff --git a/nat/src/masquerade/apalloc/alloc.rs b/nat/src/masquerade/apalloc/alloc.rs index ae7886c33f..958b0b0651 100644 --- a/nat/src/masquerade/apalloc/alloc.rs +++ b/nat/src/masquerade/apalloc/alloc.rs @@ -436,20 +436,24 @@ impl PoolBitmap { self.0.insert(index) } - pub(crate) fn add_prefix(&mut self, prefix: &Prefix, bitmap_mapping: &BTreeMap) { - match prefix { - Prefix::IPV4(p) => { - let start = p.network().to_bits(); - let end = p.last_address().to_bits(); - self.0.insert_range(start..=end); - } - Prefix::IPV6(p) => { - let start = map_address(p.network(), bitmap_mapping); - let end = map_address(p.last_address(), bitmap_mapping); - self.0.insert_range(start..=end); - } + /// Mark every address of an IPv4 prefix free. A no-op for IPv6 prefixes, whose bitmap ranges + /// come from [`PoolBitmap::add_offset_range`] instead. + pub(crate) fn add_v4_prefix(&mut self, prefix: &Prefix) { + // For IPv4 the bitmap index *is* the address, so the whole prefix always fits. + if let Prefix::IPV4(p) = prefix { + let start = p.network().to_bits(); + let end = p.last_address().to_bits(); + self.0.insert_range(start..=end); } } + + /// Mark the inclusive offset window `base..=end` free. + /// + /// Used for IPv6, where a prefix occupies a window of the offset space that may be narrower than + /// the prefix itself; see `setup::Ipv6BitmapMappings`. + pub(crate) fn add_offset_range(&mut self, base: u32, end: u32) { + self.0.insert_range(base..=end); + } } /////////////////////////////////////////////////////////////////////////////// @@ -479,12 +483,34 @@ pub(crate) fn map_offset( .map_err(|()| AllocatorError::InternalIssue("Failed to convert offset to IPv6".to_string())) } -// Reverse operation from map_offset() -pub(crate) fn map_address(address: Ipv6Addr, bitmap_mapping: &BTreeMap) -> u32 { +/// Reverse operation from [`map_offset`]. +/// +/// Fallible rather than panicking, because it is reachable with an address the *current* mapping +/// does not cover. Re-reserving a flow's allocation during a config change (see +/// `masquerade::flows::re_reserve_ip_and_port`) hands over an address allocated under the previous +/// allocator, and a config change can move or drop the prefix it came from. Returning an error +/// there invalidates one flow; panicking would abort the process. +pub(crate) fn map_address( + address: Ipv6Addr, + bitmap_mapping: &BTreeMap, +) -> Result { + let address_bits = address.to_bits(); let (prefix_start_bits, prefix_offset) = bitmap_mapping - .range(..=address.to_bits()) + .range(..=address_bits) .next_back() - .expect("This should never fail"); + .ok_or_else(|| { + AllocatorError::InternalIssue(format!("Address {address} is below every mapped prefix")) + })?; + + // The offset within the prefix must fit the bitmap: a prefix wider than the remaining budget is + // mapped with a truncated window, so an address in its tail has no index. + let offset_in_prefix = u32::try_from(address_bits - prefix_start_bits).map_err(|_| { + AllocatorError::InternalIssue(format!( + "Address {address} is beyond the mapped window of its prefix" + )) + })?; - prefix_offset + u32::try_from(address.to_bits() - prefix_start_bits).unwrap() + prefix_offset.checked_add(offset_in_prefix).ok_or_else(|| { + AllocatorError::InternalIssue(format!("Bitmap offset for {address} overflows u32")) + }) } diff --git a/nat/src/masquerade/apalloc/natip_with_bitmap.rs b/nat/src/masquerade/apalloc/natip_with_bitmap.rs index 52385055a7..73e8c04703 100644 --- a/nat/src/masquerade/apalloc/natip_with_bitmap.rs +++ b/nat/src/masquerade/apalloc/natip_with_bitmap.rs @@ -60,6 +60,6 @@ impl NatIpWithBitmap for Ipv6Addr { bitmap_mapping: &BTreeMap, ) -> Result { // Reverse operation of map_offset() - Ok(map_address(address, bitmap_mapping)) + map_address(address, bitmap_mapping) } } diff --git a/nat/src/masquerade/apalloc/setup.rs b/nat/src/masquerade/apalloc/setup.rs index 7c587a7345..58ed39e9eb 100644 --- a/nat/src/masquerade/apalloc/setup.rs +++ b/nat/src/masquerade/apalloc/setup.rs @@ -16,7 +16,7 @@ use net::ip::NextHeader; use net::packet::VpcDiscriminant; use std::collections::{BTreeMap, BTreeSet}; use std::time::Duration; -use tracing::error; +use tracing::{error, warn}; const DEFAULT_MASQUERADE_IDLE_TIMEOUT: Duration = Duration::from_mins(2); @@ -205,7 +205,7 @@ fn create_natpool( exclude_wellknown_ports: bool, ) -> NatPool { // Build mappings for IPv6 <-> u32 bitmap translation - let (bitmap_mapping, reverse_bitmap_mapping) = create_ipv6_bitmap_mappings( + let mappings = create_ipv6_bitmap_mappings( &prefixes .iter() // FIXME: Add port range, too @@ -213,12 +213,24 @@ fn create_natpool( .collect::>(), ); - // Mark all addresses as available (free) in bitmap + // Mark all addresses as available (free) in bitmap. + // + // IPv4 prefixes index by address directly. IPv6 prefixes come from the mapping's `ranges` + // rather than from the prefixes themselves, because only the mapping knows how much of each + // prefix fits in the bitmap -- deriving the range from a prefix's own size would ask the bitmap + // for offsets it does not have. let mut bitmap = PoolBitmap::new(); prefixes .iter() // FIXME: Add port range, too - .for_each(|prefix| bitmap.add_prefix(&prefix.prefix(), &reverse_bitmap_mapping)); + .map(PrefixWithOptionalPorts::prefix) + .for_each(|prefix| bitmap.add_v4_prefix(&prefix)); + for &(base, end) in &mappings.ranges { + bitmap.add_offset_range(base, end); + } + + let bitmap_mapping = mappings.forward; + let reverse_bitmap_mapping = mappings.reverse; let reserved_prefixes_ports = build_reserved_prefixes_ports(prefixes_and_ports_to_exclude_from_pools); @@ -259,44 +271,180 @@ fn prefix_bounds(prefix: &PrefixWithOptionalPorts) -> (I, I) { (addr, addr_range_end) } -// The allocator's bitmap contains u32 only. For IPv4, it maps well to the address space. For IPv6, -// we need some mapping to associate IPv6 addresses with u32 indices. This also means that we cannot -// use more than 2^32 addresses for one expose, for NAT. If the prefixes we get contain more, we'll -// just ignore the remaining addresses. Hardware limitations are such that working with 4 billion -// allocated addresses is unreallistic anyway. -#[allow(clippy::type_complexity)] -fn create_ipv6_bitmap_mappings( - prefixes: &BTreeSet, -) -> (BTreeMap, BTreeMap) { - let mut bitmap_mapping = BTreeMap::new(); - let mut reverse_bitmap_mapping = BTreeMap::new(); - let mut index = 0; +/// IPv6 prefixes, projected onto the allocator's `u32` bitmap index space. +/// +/// The bitmap indexes addresses with a `u32`. For IPv4 an address *is* its index. For IPv6 the +/// address space is far larger, so prefixes are laid end to end in a `u32` offset space, which caps +/// the total usable at `2^32` addresses across all of an expose's prefixes. Anything beyond that is +/// dropped: allocating four billion addresses is not reachable in practice, and a masquerade pool +/// does not need to offer every address of a large prefix to work. +pub(crate) struct Ipv6BitmapMappings { + /// Offset -> network address of the prefix that offset falls in. + pub(crate) forward: BTreeMap, + /// Network address of a prefix -> its base offset. + pub(crate) reverse: BTreeMap, + /// `(first offset, last offset)` inclusive, per mapped prefix. + /// + /// Carried separately because it is the *usable* window, not the prefix's own extent: the last + /// prefix to fit may be truncated, and a prefix that does not fit at all is absent entirely. + /// The bitmap must be populated from this rather than from prefix sizes. + /// + /// An inclusive end rather than a count, because a full budget is `2^32` addresses, which is one + /// more than a `u32` can hold; the last *offset* is `u32::MAX` and always fits. + pub(crate) ranges: Vec<(u32, u32)>, +} + +/// Lay `prefixes`' IPv6 members out in the bitmap's `u32` offset space. +/// +/// Truncates the address space to `u32::MAX + 1` addresses in total, dropping whole prefixes once +/// the budget is exhausted. A prefix is only ever mapped together with the number of its addresses +/// that are actually representable, so no part of the allocator can be asked for an offset outside +/// the bitmap. +fn create_ipv6_bitmap_mappings(prefixes: &BTreeSet) -> Ipv6BitmapMappings { + /// Addresses representable by a `u32` index. + const BUDGET: u128 = 1 << 32; + + let mut mappings = Ipv6BitmapMappings { + forward: BTreeMap::new(), + reverse: BTreeMap::new(), + ranges: Vec::new(), + }; + let mut index: u128 = 0; for prefix in prefixes { - if let Prefix::IPV6(p) = prefix { - let start_address = p.network().to_bits(); - bitmap_mapping.insert(index, start_address); - reverse_bitmap_mapping.insert(start_address, index); - if p.size() + u128::from(index) >= 2_u128.pow(32) { - break; - } - let Ok(psize) = u128::try_from(p.size()) else { - error!("Failed to get u128 from prefix {:#?}", p.size()); - continue; - }; - let Ok(psize_u32) = u32::try_from(psize) else { - error!("Failed to convert {psize} to u32"); - continue; - }; - index += psize_u32; + let Prefix::IPV6(p) = prefix else { continue }; + + let remaining = BUDGET - index; + if remaining == 0 { + warn!("Ran out of NAT bitmap space before prefix {p}; it will not be used"); + continue; + } + + // `PrefixSize` is not always a `u128` -- a `::/0` holds `2^128` addresses -- so treat + // anything that does not convert as "larger than the budget", which it necessarily is. + let size = u128::try_from(p.size()).unwrap_or(u128::MAX); + let usable = size.min(remaining); + if usable < size { + warn!( + "NAT bitmap space exhausted within prefix {p}: using {usable} of {size} addresses" + ); } + + // Exact by construction: `index < BUDGET` and `index + usable <= BUDGET`, so both the base + // and the inclusive end land in `0..=u32::MAX`. + let (Ok(base), Ok(end)) = (u32::try_from(index), u32::try_from(index + usable - 1)) else { + error!( + "Bitmap window {index}..={} exceeds u32; dropping prefix {p}", + index + usable - 1 + ); + continue; + }; + + let start_address = p.network().to_bits(); + mappings.forward.insert(base, start_address); + mappings.reverse.insert(start_address, base); + mappings.ranges.push((base, end)); + + index += usable; } - (bitmap_mapping, reverse_bitmap_mapping) + mappings } #[cfg(test)] mod tests { - use super::{ReserveSets, find_masquerade_portfw_overlap}; + use super::{ReserveSets, create_ipv6_bitmap_mappings, find_masquerade_portfw_overlap}; + use lpm::prefix::Prefix; + use std::collections::BTreeSet; + use std::str::FromStr; + + /// Addresses a `u32` bitmap index can address. + const BUDGET: u128 = 1 << 32; + + fn v6_set(cidrs: &[&str]) -> BTreeSet { + cidrs + .iter() + .map(|c| Prefix::from_str(c).expect("test prefixes must parse")) + .collect() + } + + /// The whole budget, and not one address more. + /// + /// A `/96` holds exactly `2^32` addresses, so its window runs to `u32::MAX` inclusive. Counting + /// addresses rather than offsets would need `2^32`, which does not fit a `u32` -- an earlier cut + /// of this fix dropped the prefix outright rather than truncating it. + #[test] + fn test_prefix_of_exactly_the_budget_fills_the_bitmap() { + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/96"])); + assert_eq!(mappings.ranges, vec![(0, u32::MAX)]); + assert_eq!(mappings.forward.len(), 1); + assert_eq!(mappings.reverse.len(), 1); + } + + /// A prefix wider than the bitmap is truncated to what fits, not dropped and not fatal. + #[test] + fn test_prefix_wider_than_the_budget_is_truncated() { + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/64"])); + assert_eq!( + mappings.ranges, + vec![(0, u32::MAX)], + "a /64 should yield the full budget, truncated" + ); + } + + /// Once the budget is gone, later prefixes are dropped rather than aliasing earlier ones. + #[test] + fn test_prefixes_after_the_budget_is_exhausted_are_dropped() { + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/64", "2001:db9::/96"])); + assert_eq!(mappings.ranges.len(), 1, "only the first prefix fits"); + assert_eq!(mappings.forward.len(), 1); + assert_eq!(mappings.reverse.len(), 1); + } + + /// Prefixes that fit are laid end to end with no gap and no overlap. + #[test] + fn test_prefixes_are_packed_contiguously() { + // Two /97s: half the budget each, so both fit exactly. + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/97", "2001:db9::/97"])); + let half = u32::try_from(BUDGET / 2).expect("half the budget fits a u32"); + assert_eq!(mappings.ranges, vec![(0, half - 1), (half, u32::MAX)]); + } + + /// Every offset a mapped window contains must round-trip back to an address, and back again. + #[test] + fn test_window_bounds_round_trip() { + use super::super::alloc::{map_address, map_offset}; + + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/112", "2001:db9::/112"])); + for &(base, end) in &mappings.ranges { + for offset in [base, end] { + let address = map_offset(offset, &mappings.forward) + .unwrap_or_else(|e| panic!("offset {offset} should map to an address: {e}")); + let back = map_address(address, &mappings.reverse) + .unwrap_or_else(|e| panic!("address {address} should map back: {e}")); + assert_eq!(offset, back, "round trip changed offset {offset}"); + } + } + } + + /// An address past a truncated prefix's window has no index, and says so rather than panicking. + /// + /// This is the path a flow allocated under a previous config takes when its address is no longer + /// representable; it must invalidate the flow, not abort the process. + #[test] + fn test_address_beyond_truncated_window_is_an_error() { + use super::super::alloc::map_address; + use std::net::Ipv6Addr; + + let mappings = create_ipv6_bitmap_mappings(&v6_set(&["2001:db8::/64"])); + // Inside the /64, but far past the first 2^32 addresses of it. + let beyond = + Ipv6Addr::from_str("2001:db8::ffff:ffff:ffff").expect("test address must parse"); + assert!( + map_address(beyond, &mappings.reverse).is_err(), + "an address outside the mapped window must not produce an index" + ); + } + use config::external::overlay::vpcpeering::VpcExpose; use lpm::prefix::{L4Protocol, PortRange, PrefixPortsSet, PrefixWithOptionalPorts}; diff --git a/nat/src/masquerade/mod.rs b/nat/src/masquerade/mod.rs index d7b2861dba..da23701c29 100644 --- a/nat/src/masquerade/mod.rs +++ b/nat/src/masquerade/mod.rs @@ -12,6 +12,7 @@ mod packet; mod protocol; mod state; mod test; +mod test_concurrency; // re exports pub use allocator_writer::MasqueradeConfig; diff --git a/nat/src/masquerade/test_concurrency.rs b/nat/src/masquerade/test_concurrency.rs new file mode 100644 index 0000000000..1a4c445330 --- /dev/null +++ b/nat/src/masquerade/test_concurrency.rs @@ -0,0 +1,366 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +//! Reconfiguration of the masquerade allocator, raced against the data path. +//! +//! Everything else in this crate's test suite either exercises the allocator on one thread or +//! exercises concurrent allocation against a *fixed* allocator (`apalloc::test_alloc`'s +//! `concurrency_tests`). Neither covers the operation this module is about: replacing the +//! allocator underneath a running data path. +//! +//! [`NatAllocatorWriter::update_nat_allocator`] is not a table swap. It walks the *live* flow +//! table, re-reserving each active flow's `(ip, port)` in a freshly built allocator, and only then +//! publishes it. Two structures are therefore read-modify-written with no single atomic step +//! covering both, so the question is whether the lock discipline around that window is airtight. +//! +//! It is tighter than it first looks. The migration walk holds a [`FlowTableReadGuard`] across the +//! publish, and flow *insertion* takes the same lock exclusively, so no flow can appear between the +//! walk and the store. What that guard does *not* cover is an already-present flow acquiring its +//! allocation late: per-flow NAT state sits behind the flow's own lock, not the table's. A flow the +//! walk skipped for having no allocation yet could take one from the outgoing allocator afterwards, +//! and the incoming allocator would never have reserved it. +//! +//! Whether the data path can actually reach that ordering is what these tests are for. The +//! invariant is the one that matters either way, and does not depend on the answer: **no two active +//! flows may hold the same masquerading allocation.** Two flows translated to the same address and +//! port are indistinguishable on the wire, so the reply to one can be delivered to the other. +//! +//! Configs come from `k8s-intf`'s generators rather than being hand-written, so the shapes fed to +//! the allocator are the shapes the CRD can actually express. `bolero` supplies the shape axis. +//! +//! # Why this is not a shuttle test +//! +//! The intent was `bolero` for shapes and shuttle for interleavings, as +//! `concurrency/tests/quiescent_shuttle.rs` does. That is not reachable here: `FlowTable::insert` +//! starts a per-flow expiry timer with `tokio::task::spawn`, so every path that creates a flow needs +//! a live tokio runtime. Shuttle replaces the primitives behind `concurrency::sync` and +//! `concurrency::thread` with cooperatively scheduled ones; it has no view into tokio's scheduler or +//! timer wheel, and a tokio runtime inside a shuttle execution is not a supported combination. +//! +//! So anything downstream of flow insertion is currently outside what a model checker can drive, and +//! that is a property of the flow table rather than of this test. What is left is real threads on a +//! tokio runtime: the OS picks the interleaving, so a narrow window is reached by luck rather than by +//! construction. That still earns its keep -- it is a regression test for the invariant, and under +//! `ThreadSanitizer` it becomes a race detector over generated config shapes -- but it does not *prove* +//! the window is closed, and should not be read as doing so. +//! +//! Making this model-checkable means giving the flow table a way to insert without arming a timer +//! (injecting the timer as a dependency, or a `cfg` seam), at which point this test can move to the +//! shuttle pattern unchanged. + +#![cfg(test)] + +use ahash::HashMap; +use concurrency::sync::Arc; +use concurrency::thread; +use config::external::overlay::ValidatedOverlay; +use config::{ExternalConfig, GenId}; +use flow_entry::flow_table::{FlowLookup, FlowTable}; +use k8s_intf::bolero::LegalValue; +use k8s_intf::gateway_agent_crd::GatewayAgent; +use net::buffer::TestBuffer; +use net::flows::flow_info_item::ExtractRef; +use net::packet::test_utils::build_test_tcp_ipv4_packet; +use net::packet::{Packet, VpcDiscriminant}; +use pipeline::{DynPipeline, NetworkFunction}; +use std::collections::BTreeSet; +use std::net::IpAddr; + +use crate::masquerade::allocator_writer::{NatAllocatorReader, NatAllocatorReaderFactory}; +use crate::masquerade::state::MasqueradeState; +use crate::masquerade::{MasqueradeConfig, NatAllocatorWriter}; +use crate::{IcmpErrorHandler, Masquerade}; + +/// Stand-in for the flow-filter stage, which is not what is under test here. +/// +/// Mirrors `masquerade::test::TestFlowFilter`; duplicated rather than shared because that module is +/// `#![cfg(test)]` and its helper is private to it. +#[derive(Default)] +struct TestFlowFilter(HashMap); + +impl TestFlowFilter { + fn with_peerings(peerings: Vec<(VpcDiscriminant, VpcDiscriminant)>) -> Self { + let mut new = TestFlowFilter::default(); + for (src_vpcd, dst_vpcd) in peerings { + new.0.insert(src_vpcd, dst_vpcd); + new.0.insert(dst_vpcd, src_vpcd); + } + new + } +} + +impl NetworkFunction for TestFlowFilter { + fn process<'a, Input: Iterator> + 'a>( + &'a mut self, + input: Input, + ) -> impl Iterator> + 'a { + input.filter_map(|mut packet| { + let src_vpcd = packet.meta().src_vpcd?; + // Unpeered source: drop rather than panic. A generated config need not peer every VPC, + // and an unroutable packet is not what this test is about. + let dst_vpcd = *self.0.get(&src_vpcd)?; + packet.meta_mut().dst_vpcd = Some(dst_vpcd); + Some(packet) + }) + } +} + +/// One masquerading peering drawn from a config, with an address inside its private range. +struct MasqTarget { + src_vpcd: VpcDiscriminant, + src_ip: IpAddr, +} + +/// Every masquerading peering in `config`, paired with a source address it will translate. +/// +/// Returns an empty vector when the config masquerades nothing, in which case there is no allocator +/// to race against and the caller should skip. +fn masquerading_targets(config: &MasqueradeConfig) -> Vec { + let mut targets = Vec::new(); + for peering in config.iter() { + for expose in peering.peering.local().valexp() { + let Some(nat) = expose.nat() else { continue }; + if !nat.is_masquerade() { + continue; + } + // The first address of the first exposed prefix is inside what this peering translates, + // which is all the data path needs to take an allocation. + // + // IPv4 only, but for the packet builder rather than the allocator: `drive_data_path` + // builds IPv4 TCP packets. Wide IPv6 masquerade ranges reach the allocator either way, + // via `setup` building the pool for every peering in the config. + let Some(prefix) = expose.ips().iter().next() else { + continue; + }; + let address = prefix.prefix().as_address(); + if address.is_ipv4() { + targets.push(MasqTarget { + src_vpcd: peering.src_vpcd, + src_ip: address, + }); + break; + } + } + } + targets +} + +/// A `DynPipeline` holds boxed stages without a `Send` bound, so it cannot cross a thread +/// boundary. Each thread therefore builds its own; the state that matters -- the flow table and the +/// published allocator -- is shared, and a pipeline holds none of it. +fn make_pipeline( + flow_table: &Arc, + alloc_reader: NatAllocatorReader, + peerings: Vec<(VpcDiscriminant, VpcDiscriminant)>, +) -> DynPipeline { + DynPipeline::new() + .add_stage(IcmpErrorHandler::new(flow_table.clone())) + .add_stage(FlowLookup::new("flow-lookup", flow_table.clone())) + .add_stage(TestFlowFilter::with_peerings(peerings)) + .add_stage(Masquerade::new("masq", flow_table.clone(), alloc_reader)) +} + +/// The peerings a generated config masquerades, in the form the flow filter wants. +fn peering_pairs(config: &MasqueradeConfig) -> Vec<(VpcDiscriminant, VpcDiscriminant)> { + config.iter().map(|p| (p.src_vpcd, p.dst_vpcd)).collect() +} + +/// Install the initial allocator and hand back what both threads need. +fn setup( + overlay: &ValidatedOverlay, + genid: GenId, +) -> ( + Arc, + NatAllocatorWriter, + NatAllocatorReaderFactory, + MasqueradeConfig, +) { + let nat_config = MasqueradeConfig::new(overlay.vpc_table(), genid); + let mut alloc_writer = NatAllocatorWriter::new(); + let reader_factory = alloc_writer.get_reader_factory(); + let flow_table = Arc::new(FlowTable::default()); + + alloc_writer.update_nat_allocator(nat_config.clone(), &flow_table); + (flow_table, alloc_writer, reader_factory, nat_config) +} + +/// Push one TCP packet per target through the pipeline, taking an allocation for each. +fn drive_data_path( + pipeline: &mut DynPipeline, + targets: &[MasqTarget], + sport_base: u16, +) { + for (index, target) in targets.iter().enumerate() { + // Distinct source ports so each packet is a distinct flow rather than a hit on the + // previous one. + let sport = sport_base.wrapping_add(u16::try_from(index).unwrap_or(0)) | 1; + let mut packet = build_test_tcp_ipv4_packet( + &target.src_ip.to_string(), + // Any destination: the flow filter decides the peer, not the address. + "203.0.113.9", + sport, + 443, + ); + packet.meta_mut().set_overlay(true); + packet.meta_mut().set_masquerade(true); + packet.meta_mut().src_vpcd = Some(target.src_vpcd); + let _out: Vec<_> = pipeline.process(std::iter::once(packet)).collect(); + } +} + +/// The invariant: no two active flows may hold the same masquerading allocation. +/// +/// A duplicate means two distinct flows translate to the same address and port, so replies cannot +/// be attributed to the right one. +fn assert_no_duplicate_allocations(flow_table: &FlowTable) { + let mut seen: BTreeSet<(IpAddr, u16)> = BTreeSet::new(); + let mut duplicates = Vec::new(); + + let guard = flow_table.for_each_flow_filtered( + |_key, flow_info| flow_info.is_active(), + |key, flow_info| { + let locked = flow_info.locked.read(); + let Some(state) = locked.nat_state.extract_ref::() else { + return; + }; + let Some(alloc) = state.allocation() else { + return; + }; + let entry = (alloc.ip(), alloc.port().as_u16()); + if !seen.insert(entry) { + duplicates.push(format!("{key} -> {}:{}", entry.0, entry.1)); + } + }, + ); + drop(guard); + + assert!( + duplicates.is_empty(), + "two active flows share a masquerading allocation: {duplicates:?}" + ); +} + +/// Race a reconfiguration against the data path, then check the invariant. +/// +/// The second config reuses the first's peerings under a fresh generation id. That is deliberate: +/// an unchanged config takes `update_nat_allocator`'s early-return path and a wholly different one +/// invalidates every flow, so neither reaches the re-reservation walk. Bumping only the generation +/// makes every live flow a migration candidate, which is the widest form of the walk. +fn run_reconfiguration_race(agent: &GatewayAgent) { + let Ok(external) = ExternalConfig::try_from(agent) else { + return; + }; + let Ok(validated) = external.validate() else { + return; + }; + let overlay = validated.external().overlay(); + + let genid = validated.genid(); + let (flow_table, mut alloc_writer, reader_factory, config) = setup(overlay, genid); + let targets = masquerading_targets(&config); + if targets.is_empty() { + // Nothing masquerades, so there is no allocator to migrate and no second thread's worth of + // work. Shuttle's PCT scheduler panics on a body without real concurrency, so skip. + return; + } + let pairs = peering_pairs(&config); + + // Seed flows *before* the race so the migration walk has something to re-reserve. + { + let mut pipeline = make_pipeline(&flow_table, reader_factory.handle(), pairs.clone()); + drive_data_path(&mut pipeline, &targets, 1000); + } + + let next_config = MasqueradeConfig::new(overlay.vpc_table(), genid + 1); + let flow_table_for_writer = flow_table.clone(); + let flow_table_for_data_path = flow_table.clone(); + + // Both threads touch the flow table, whose insert path arms a tokio timer, so both need the + // runtime in scope. + let handle = tokio::runtime::Handle::current(); + let writer_handle = handle.clone(); + + let reconfigure = thread::spawn(move || { + let _guard = writer_handle.enter(); + alloc_writer.update_nat_allocator(next_config, &flow_table_for_writer); + }); + let data_path = thread::spawn(move || { + let _guard = handle.enter(); + // Fresh source ports: these flows are created while the swap is in flight, which is the + // window the guard does not obviously cover. + let mut pipeline = make_pipeline(&flow_table_for_data_path, reader_factory.handle(), pairs); + drive_data_path(&mut pipeline, &targets, 2000); + }); + + reconfigure.join().expect("reconfigure thread panicked"); + data_path.join().expect("data path thread panicked"); + + assert_no_duplicate_allocations(&flow_table); +} + +/// Race a reconfiguration against the data path across many generated config shapes. +/// +/// Not gated on `shuttle`: see the module docs for why a model checker cannot drive this path. A +/// `shuttle` build compiles and runs this as ordinary threads, which is harmless but proves nothing +/// extra, so there is no separate shuttle entry point to imply otherwise. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reconfiguration_race_preserves_allocation_uniqueness() { + bolero::check!() + .with_type::>() + .for_each(|agent| run_reconfiguration_race(agent.as_ref())); +} + +/// A validated config with an IPv6 masquerade range wider than the bitmap can index must build. +/// +/// The allocator indexes a masquerade range's addresses with a `u32`, so an IPv6 prefix shorter than +/// a `/96` holds more addresses than it can represent. `create_ipv6_bitmap_mappings` truncates such +/// a prefix to the addresses that fit, which is what the allocator has always documented; before +/// that truncation was implemented, building the pool panicked in `map_address` on an offset that +/// did not fit a `u32`. +/// +/// Validation does not reject these ranges, so this is reachable straight from operator input: a +/// `GatewayAgent` carrying an IPv6 masquerade `as` range validates, and under the shipped +/// `panic = "abort"` the resulting panic took the whole process down. +/// +/// No concurrency and no packets: building the allocator is enough. +#[test] +fn test_wide_ipv6_masquerade_range_is_truncated_not_fatal() { + use config::external::overlay::Overlay; + use config::external::overlay::vpc::{Vpc, VpcTable}; + use config::external::overlay::vpcpeering::{ + VpcExpose, VpcManifest, VpcPeering, VpcPeeringTable, + }; + + // A /64 on each side: same family, no port ranges, no exclusions -- everything + // `VpcExpose::validate` asks of a masquerading expose. + let masq = VpcExpose::empty() + .make_masquerade(None) + .expect("masquerade is a legal mode") + .ip("2001:db8:1::/64".into()) + .as_range("2001:db8:2::/64".into()) + .expect("as_range is legal with NAT configured"); + let plain = VpcExpose::empty().ip("2001:db8:3::/64".into()); + + let mut left = VpcManifest::new("VPC-1"); + left.add_expose(masq); + let mut right = VpcManifest::new("VPC-2"); + right.add_expose(plain); + + let mut vpcs = VpcTable::new(); + vpcs.add(Vpc::new("VPC-1", "aaaaa", 100).expect("legal vpc")) + .expect("first vpc"); + vpcs.add(Vpc::new("VPC-2", "bbbbb", 200).expect("legal vpc")) + .expect("second vpc"); + + let mut peerings = VpcPeeringTable::new(); + peerings + .add(VpcPeering::with_default_group("peering-1", left, right)) + .expect("first peering"); + + let overlay = Overlay::new(vpcs, peerings) + .validate() + .expect("a v6 masquerade expose is accepted by validation -- that is the point"); + + // Used to panic in `map_address` while building the bitmap. + let (_flow_table, _writer, _factory, _config) = setup(&overlay, 1); +}