diff --git a/acl-filter/src/tests.rs b/acl-filter/src/tests.rs index eab77405b8..b545c7bb12 100644 --- a/acl-filter/src/tests.rs +++ b/acl-filter/src/tests.rs @@ -949,7 +949,6 @@ mod end_to_end { "port-forwarder", portfw_writer.reader(), flow_table.clone(), - allocator.get_reader(), )); // Masquerade (creates the related flow pair used by 'flow'-scoped replies) diff --git a/config/src/external/overlay/vpc.rs b/config/src/external/overlay/vpc.rs index e05c31267a..5225554b9b 100644 --- a/config/src/external/overlay/vpc.rs +++ b/config/src/external/overlay/vpc.rs @@ -310,7 +310,7 @@ impl Vpc { .map(Peering::validate) .collect::>()?; - let route_table = VpcRouteTable::build(&validated_peerings).validate()?; + let route_table = VpcRouteTable::build(&validated_peerings)?; let validated_vpc = ValidatedVpc { name: self.name.clone(), diff --git a/config/src/external/overlay/vpcrouting.rs b/config/src/external/overlay/vpcrouting.rs index 3f7ed19ac7..c267c428a3 100644 --- a/config/src/external/overlay/vpcrouting.rs +++ b/config/src/external/overlay/vpcrouting.rs @@ -125,9 +125,13 @@ impl VpcRouteTable { self.table.values().flat_map(VpcRouteSet::iter) } - #[must_use] /// Build a `VpcRouteTable` from the set of `ValidatedPeering` of a VPC - pub fn build(peerings: &Vec) -> Self { + /// + /// # Errors + /// + /// This function returns `ConfigError` if the `VpcRouteTable` does not + /// pass validation successfully + pub fn build(peerings: &Vec) -> Result { let mut rt = VpcRouteTable::new(); for peering in peerings { for expose in peering.remote().valexp() { @@ -150,7 +154,7 @@ impl VpcRouteTable { } } } - rt + rt.validate() } /// Consume and validate a `VpcRouteTable` @@ -162,7 +166,7 @@ impl VpcRouteTable { /// 2) destinations cannot overlap except if they are masqueraded or a default /// 3) overlapping destinations, when allowed, must use the same gateway group /// - pub fn validate(self) -> Result { + fn validate(self) -> Result { let all_routes: Vec<&VpcRoute> = self.table.values().flat_map(VpcRouteSet::iter).collect(); for (i, &route) in all_routes.iter().enumerate() { for &other in &all_routes[i + 1..] { diff --git a/dataplane/src/packet_processor/mod.rs b/dataplane/src/packet_processor/mod.rs index 6ca5d04d3c..4e9190c5ce 100644 --- a/dataplane/src/packet_processor/mod.rs +++ b/dataplane/src/packet_processor/mod.rs @@ -122,7 +122,6 @@ pub(crate) fn start_router( "port-forwarder", portfw_factory.handle(), flow_table_clone.clone(), - natallocator_factory.handle(), ); let pkt_stats_nf = PacketStatsNF::new(pkt_stats.clone()); diff --git a/flow-entry/src/flow_table/table.rs b/flow-entry/src/flow_table/table.rs index 3dc0922bf1..27695322f7 100644 --- a/flow-entry/src/flow_table/table.rs +++ b/flow-entry/src/flow_table/table.rs @@ -40,10 +40,15 @@ fn hasher_state() -> &'static RandomState { HASHER_STATE.get_or_init(|| RandomState::with_seeds(0, 0, 0, 0)) } -/// A read guard to the `FlowTable`. While a guard like this one exists, other threads -/// attempting insertions in the corresponding `FlowTable` will wait. This guard is -/// returned by some methods that iterate over the `FlowTable` as a knob to allow callers -/// to block the table from insertions, without exposing the internal types +/// A read guard to the `FlowTable`, returned by the methods that iterate over it so a caller can +/// keep holding what the iteration held, without exposing the internal types. +/// +/// It excludes only the operations that take the table for writing, which today means +/// [`FlowTable::reshard`]. It does **not** hold off insertion or removal: those take the same read +/// lock and mutate the map through it, so they proceed alongside this guard. A caller that must not +/// miss a flow inserted while it works cannot get that from this guard; it needs the inserting side +/// to re-check its own work afterwards, the way masquerade re-checks the allocator a flow was built +/// from once the flow is in the table. pub struct FlowTableReadGuard<'a>( #[allow(unused)] RwLockReadGuard<'a, DashMap, RandomState>>, ); diff --git a/nat/src/masquerade/allocator_writer.rs b/nat/src/masquerade/allocator_writer.rs index 3a0e2789ed..6be0ece775 100644 --- a/nat/src/masquerade/allocator_writer.rs +++ b/nat/src/masquerade/allocator_writer.rs @@ -11,8 +11,8 @@ use flow_entry::flow_table::FlowTable; use net::packet::VpcDiscriminant; use tracing::debug; -use crate::masquerade::flows::reconcile_nat_flows; -use crate::masquerade::flows::remove_allocator_from_flows; +use crate::masquerade::flows::check_masquerading_flows; +use crate::masquerade::flows::invalidate_masquerade_flows; use crate::masquerade::flows::upgrade_all_masquerading_flows; #[derive(Debug, PartialEq, Clone)] @@ -21,6 +21,19 @@ pub(crate) struct MasqueradePeering { pub(crate) dst_vpcd: VpcDiscriminant, pub(crate) peering: ValidatedPeering, } + +impl MasqueradePeering { + /// Tells if this peering masquerades locally, i.e. it has at least one local expose configured + /// with masquerading. N.B. `MasqueradeConfig` also holds peerings that only port-forward. + pub(crate) fn has_masquerade(&self) -> bool { + self.peering + .local() + .valexp() + .iter() + .any(ValidatedExpose::has_masquerade) + } +} + #[derive(Debug, Default, Clone, PartialEq)] pub struct MasqueradeConfig { peerings: Vec, @@ -62,22 +75,19 @@ impl MasqueradeConfig { } pub(crate) fn has_masquerading_peerings(&self) -> bool { - self.peerings.iter().map(|p| &p.peering).any(|p| { - p.local() - .valexp() - .iter() - .any(ValidatedExpose::has_masquerade) - }) + self.peerings.iter().any(MasqueradePeering::has_masquerade) } - pub(crate) fn get_peering( + /// Find a peering between two VPCs identified by their `VpcDiscriminant`s and + /// that has a local masqueraded expose + pub(crate) fn find_masquerade_peering( &self, src_vpcd: VpcDiscriminant, dst_vpcd: VpcDiscriminant, ) -> Option<&MasqueradePeering> { self.peerings .iter() - .find(|p| p.src_vpcd == src_vpcd && p.dst_vpcd == dst_vpcd) + .find(|p| p.src_vpcd == src_vpcd && p.dst_vpcd == dst_vpcd && p.has_masquerade()) } } @@ -102,8 +112,11 @@ impl NatAllocatorWriter { /// Install the allocator for a new NAT configuration. /// - /// Removing masquerade invalidates NAT flows and clears port-forward leases. Replacement - /// carries compatible allocations forward. + /// An unchanged NAT configuration keeps the current allocator, and only advances the generation + /// id (and that of the flows). A configuration that no longer masquerades drops the allocator and + /// invalidates the flows that were using it. Any other change installs a replacement, which + /// carries over the allocations it can still honour and invalidates the flows whose tuples it + /// cannot serve. pub fn update_nat_allocator( &mut self, nat_config: MasqueradeConfig, @@ -112,7 +125,7 @@ impl NatAllocatorWriter { ) { let curr_allocator = self.0.load_full(); - // keep state as-is if config did not change, and just upgrade flows + // keep state as-is if config did not change, and just upgrade flows' gen id if let Some(current) = curr_allocator.as_ref() && current.config() == &nat_config { @@ -122,19 +135,20 @@ impl NatAllocatorWriter { return; } - // if we transition to a config without masquerading, flush allocator and remove all flows + // if we transition to a config without masquerading, remove allocator and masquerading flows if !nat_config.has_masquerading_peerings() { if curr_allocator.is_some() { - debug!("Removing masquerade allocator and its flow state"); + debug!("Removing NAT allocator"); self.0.store(None); - remove_allocator_from_flows(flow_table); + invalidate_masquerade_flows(flow_table); } return; } let allocator = NatAllocator::new(nat_config, genid); - let guard = reconcile_nat_flows(flow_table, &allocator); - debug!("Installing masquerade NAT allocator..."); + let guard = check_masquerading_flows(flow_table, &allocator); + + debug!("Installing NAT allocator (genid {}).", allocator.genid()); self.0.store(Some(Arc::new(allocator))); drop(guard); } diff --git a/nat/src/masquerade/apalloc/alloc.rs b/nat/src/masquerade/apalloc/alloc.rs index fbfd8f5cdf..51cf9caf5b 100644 --- a/nat/src/masquerade/apalloc/alloc.rs +++ b/nat/src/masquerade/apalloc/alloc.rs @@ -4,12 +4,14 @@ //! Masquerade IP allocation. See the architecture diagram in `mod.rs`. use super::region::AddrInterval; +use super::reserved::{ReservedForAddr, ReservedPorts}; use super::{NatIpWithBitmap, port_alloc}; use crate::masquerade::allocation::AllocatorError; use crate::masquerade::natip::NatIp; use crate::port::NatPort; use crate::ranges::IpRange; use concurrency::sync::{Arc, RwLock, RwLockReadGuard, Weak}; +use port_alloc::PortAllocator; use roaring::RoaringBitmap; use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::net::{IpAddr, Ipv6Addr}; @@ -256,12 +258,14 @@ impl AllocatedIp { fn new( ip: I, ip_allocator: IpAllocator, + reserved: ReservedForAddr, randomize: bool, exclude_wellknown_ports: bool, ) -> Self { + let port_allocator = PortAllocator::new(reserved, randomize, exclude_wellknown_ports); Self { ip, - port_allocator: port_alloc::PortAllocator::new(randomize, exclude_wellknown_ports), + port_allocator, ip_allocator, } } @@ -270,6 +274,14 @@ impl AllocatedIp { self.ip } + /// The ports of this address that masquerade may not hand out. + /// + /// A port block reads this off the address it belongs to when it is created, so every path that + /// creates one honours the reservations without having to be told about them. + pub(crate) fn reserved_ports(&self) -> &ReservedForAddr { + self.port_allocator.reserved_ports() + } + // Used for Display; should probably not be accessed directly anywhere else pub(crate) fn port_allocator(&self) -> &port_alloc::PortAllocator { &self.port_allocator @@ -326,12 +338,19 @@ pub(crate) struct NatPool { bitmap_mapping: BTreeMap, reverse_bitmap_mapping: BTreeMap, in_use: VecDeque>>, + /// The public tuples of this region that port forwarding may claim. Applied to every address as + /// it is put to use, so masquerade never hands one of them out. + reserved: ReservedPorts, exclude_wellknown_ports: bool, } impl NatPool { /// Build a pool over one disjoint public region. - pub(crate) fn for_range(range: AddrInterval, exclude_wellknown_ports: bool) -> Self { + pub(crate) fn for_range( + range: AddrInterval, + reserved: ReservedPorts, + exclude_wellknown_ports: bool, + ) -> Self { // IPv6 uses offsets from the region start because its addresses do not fit in the bitmap. let bitmap_mapping = BTreeMap::from([(0u32, range.start)]); let reverse_bitmap_mapping = BTreeMap::from([(range.start, 0u32)]); @@ -352,6 +371,7 @@ impl NatPool { bitmap_mapping, reverse_bitmap_mapping, in_use: VecDeque::new(), + reserved, exclude_wellknown_ports, } } @@ -384,11 +404,16 @@ impl NatPool { // Retrieve the first available offset let offset = self.bitmap.pop_ip()?; + // the ip being allocated let ip = I::try_from_offset(offset, &self.bitmap_mapping)?; + // determine the set of reserved ports that cannot be allocated for this ip + let reserved = self.reserved.for_addr(ip.to_ip_addr()); + Ok(AllocatedIp::new( ip, ip_allocator, + reserved, randomize, self.exclude_wellknown_ports, )) @@ -438,9 +463,13 @@ impl NatPool { // drops an AllocatedIp and its reference count goes to 0, but it hasn't called the drop() // function to remove the IP from the bitmap in that other thread yet). let _ = self.bitmap.set_ip_allocated(offset); + + // Reservations apply here as well: an address put to use by carrying a live masquerade flow + // over to a new allocator must not be able to re-take a port that port forwarding claims. let arc_ip = Arc::new(AllocatedIp::new( ip, ip_allocator, + self.reserved.for_addr(ip.to_ip_addr()), randomize, self.exclude_wellknown_ports, )); diff --git a/nat/src/masquerade/apalloc/concurrent_fuzz.rs b/nat/src/masquerade/apalloc/concurrent_fuzz.rs index f4caa2ffd8..bbf4920ec8 100644 --- a/nat/src/masquerade/apalloc/concurrent_fuzz.rs +++ b/nat/src/masquerade/apalloc/concurrent_fuzz.rs @@ -192,16 +192,18 @@ impl Scenario { fn specs(&self) -> Vec { self.ranges .iter() - .map(|ranges| PoolSpec { - public_ranges: ranges - .iter() - .map(|&(offset, length)| { - let start = u128::from(offset); - let end = (start + u128::from(length) - 1).min(WINDOW - 1); - AddrInterval::new(BASE + start, BASE + end) - }) - .collect(), - idle_timeout: IDLE_TIMEOUT, + .map(|ranges| { + PoolSpec::new( + ranges + .iter() + .map(|&(offset, length)| { + let start = u128::from(offset); + let end = (start + u128::from(length) - 1).min(WINDOW - 1); + AddrInterval::new(BASE + start, BASE + end) + }) + .collect(), + IDLE_TIMEOUT, + ) }) .collect() } @@ -389,11 +391,11 @@ fn stress_test_config_change() { #[concurrency::model_test] fn printing_the_pool_does_not_wedge_it_against_a_flow_ending() { concurrency::stress(|| { - let specs = vec![PoolSpec { - // One address, so the flow that ends is the last holder of the one being printed. - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + // One address, so the flow that ends is the last holder of the one being printed. + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pools = Arc::new(pool_sets_for_specs::( &specs, NextHeader::TCP, @@ -419,10 +421,10 @@ fn printing_the_pool_does_not_wedge_it_against_a_flow_ending() { #[concurrency::model_test] fn reservation_racing_block_release_is_not_an_internal_error() { concurrency::stress(|| { - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pools = Arc::new(pool_sets_for_specs::( &specs, NextHeader::TCP, @@ -452,10 +454,10 @@ fn reservation_racing_block_release_is_not_an_internal_error() { fn tidying_a_dead_block_entry_does_not_drop_a_live_one() { concurrency::stress(|| { let address = Ipv4Addr::from(u32::try_from(BASE).unwrap_or_else(|_| unreachable!())); - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pools = Arc::new(pool_sets_for_specs::( &specs, NextHeader::TCP, diff --git a/nat/src/masquerade/apalloc/display.rs b/nat/src/masquerade/apalloc/display.rs index 4298d3d056..c0b96e7e06 100644 --- a/nat/src/masquerade/apalloc/display.rs +++ b/nat/src/masquerade/apalloc/display.rs @@ -24,7 +24,7 @@ impl Display for NatAllocator { fn fmt(&self, f: &mut Formatter<'_>) -> Result { Heading("Masquerade NAT allocator table").fmt(f)?; - writeln!(f, "randomize: {}", self.randomize)?; + writeln!(f, "randomize: {}", self.config.randomize())?; writeln!(f, "source pools (IPv4):")?; writeln!(with_indent!(f), "{}", self.pools_src44)?; diff --git a/nat/src/masquerade/apalloc/mod.rs b/nat/src/masquerade/apalloc/mod.rs index 8343dbe756..05b9204473 100644 --- a/nat/src/masquerade/apalloc/mod.rs +++ b/nat/src/masquerade/apalloc/mod.rs @@ -75,8 +75,7 @@ use crate::NatPort; use crate::masquerade::MasqueradeConfig; pub use crate::masquerade::apalloc::natip_with_bitmap::NatIpWithBitmap; use crate::masquerade::natip::NatIp; -use concurrency::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; -use concurrency::sync::{Arc, RwLock, Weak}; +use concurrency::sync::atomic::{AtomicI64, Ordering}; use config::GenId; use net::ip::NextHeader; use net::packet::VpcDiscriminant; @@ -95,6 +94,7 @@ mod natip_with_bitmap; mod pool_fuzz; mod port_alloc; mod region; +mod reserved; mod setup; mod test_alloc; @@ -139,7 +139,7 @@ impl PoolTableKey { /////////////////////////////////////////////////////////////////////////////// #[derive(Debug)] -struct PoolTable( +pub(crate) struct PoolTable( BTreeMap, alloc::PoolSet>, ); @@ -199,20 +199,6 @@ impl PoolTable { } self.0.insert(key, pool_set); } - - fn public_allocator( - &self, - protocol: NextHeader, - dst_vpcd: VpcDiscriminant, - addr: J, - ) -> Option<&alloc::IpAllocator> { - self.0 - .iter() - .filter(|(key, _)| key.protocol == protocol && key.dst_vpcd == dst_vpcd) - .flat_map(|(_, pools)| pools.regions()) - .find(|region| region.range().contains(addr.to_addr_bits())) - .map(alloc::PoolRegion::allocator) - } } /////////////////////////////////////////////////////////////////////////////// @@ -253,14 +239,6 @@ impl Display for Allocation { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -struct PortForwardLeaseKey { - protocol: NextHeader, - peer_vpcd: VpcDiscriminant, - ip: IpAddr, - port: u16, -} - /// [`NatAllocator`] is the IP addresses and ports allocator for masquerade. /// /// Internally, it contains various bitmap-based IP pools, and each IP address allocated from these @@ -272,27 +250,20 @@ pub struct NatAllocator { genid: AtomicI64, pools_src44: PoolTable, pools_src66: PoolTable, - port_forward_leases: RwLock>>, - port_forward_lease_uses: AtomicUsize, - randomize: bool, } impl NatAllocator { #[must_use] pub(crate) fn new(config: MasqueradeConfig, genid: GenId) -> Self { debug!("Building NAT allocator for genid {genid}"); - let mut allocator = Self { - config: MasqueradeConfig::default(), + let pools_src44 = NatAllocator::build_pool44(&config); + let pools_src66 = NatAllocator::build_pool66(&config); + Self { + config, genid: AtomicI64::new(genid), - pools_src44: PoolTable::new(), - pools_src66: PoolTable::new(), - port_forward_leases: RwLock::new(BTreeMap::new()), - port_forward_lease_uses: AtomicUsize::new(0), - randomize: config.randomize(), - }; - allocator.build_pools(&config); - allocator.config = config; - allocator + pools_src44, + pools_src66, + } } pub(crate) fn config(&self) -> &MasqueradeConfig { @@ -310,68 +281,6 @@ impl NatAllocator { self.genid.store(genid, Ordering::Relaxed); } - /// Reserve a public tuple while port-forwarded flows use it. - /// - /// Several clients may use one forwarding rule, so they share one lease. A tuple outside the - /// masquerade pools, or in the well-known range masquerade already excludes, needs no lease. - pub(crate) fn reserve_port_forward( - &self, - protocol: NextHeader, - peer_vpcd: VpcDiscriminant, - ip: IpAddr, - port: std::num::NonZero, - ) -> Result>, AllocatorError> { - if !matches!(protocol, NextHeader::TCP | NextHeader::UDP) { - return Err(AllocatorError::UnsupportedProtocol(protocol)); - } - if port.get() < port_alloc::IANA_WELLKNOWN_PORT_LIMIT { - return Ok(None); - } - - let key = PortForwardLeaseKey { - protocol, - peer_vpcd, - ip, - port: port.get(), - }; - let mut leases = self.port_forward_leases.write(); - if self - .port_forward_lease_uses - .fetch_add(1, Ordering::Relaxed) - .is_multiple_of(256) - { - leases.retain(|_, lease| lease.upgrade().is_some()); - } - if let Some(existing) = leases.get(&key) { - match existing.upgrade() { - Some(lease) => return Ok(Some(lease)), - // Reservations may stop before the next stale-entry sweep. - None => { - leases.remove(&key); - } - } - } - - let nat_port = NatPort::new_port(port); - let allocation = match ip { - IpAddr::V4(ip) => { - let Some(pool) = self.pools_src44.public_allocator(protocol, peer_vpcd, ip) else { - return Ok(None); - }; - Allocation::V4(pool.reserve(ip, nat_port)?) - } - IpAddr::V6(ip) => { - let Some(pool) = self.pools_src66.public_allocator(protocol, peer_vpcd, ip) else { - return Ok(None); - }; - Allocation::V6(pool.reserve(ip, nat_port)?) - } - }; - let lease = Arc::new(allocation); - leases.insert(key, Arc::downgrade(&lease)); - Ok(Some(lease)) - } - fn allocate_v4( &self, src_vpcd: VpcDiscriminant, diff --git a/nat/src/masquerade/apalloc/pool_fuzz.rs b/nat/src/masquerade/apalloc/pool_fuzz.rs index 888695c1f4..9f7ee7cc08 100644 --- a/nat/src/masquerade/apalloc/pool_fuzz.rs +++ b/nat/src/masquerade/apalloc/pool_fuzz.rs @@ -14,6 +14,7 @@ use super::setup::{PoolSpec, pool_sets_for_specs}; use crate::masquerade::allocation::AllocatorError; use crate::port::NatPort; use bolero::{Driver, TypeGenerator}; +use lpm::prefix::{PortRange, PrefixPortsSet, PrefixWithOptionalPorts}; use net::ip::NextHeader; use std::collections::{BTreeMap, BTreeSet}; use std::net::{Ipv4Addr, Ipv6Addr}; @@ -74,10 +75,7 @@ impl Config { fn specs(&self) -> Vec { self.owner_ranges() .into_iter() - .map(|public_ranges| PoolSpec { - public_ranges, - idle_timeout: IDLE_TIMEOUT, - }) + .map(|public_ranges| PoolSpec::new(public_ranges, IDLE_TIMEOUT)) .collect() } @@ -179,10 +177,10 @@ fn freed_allocations_become_available_again() { #[test] fn a_port_freed_while_neighbours_are_held_is_reused() { - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let mut held: Vec<_> = (0..5) @@ -267,10 +265,10 @@ fn a_region_can_be_allocated_dry() { const PORTS_PER_ADDRESS: usize = 65536 - 1024; const ADDRESSES: usize = 2; - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE + ADDRESSES as u128 - 1)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE + ADDRESSES as u128 - 1)], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let first = Ipv4Addr::from(u32::try_from(BASE).unwrap_or_else(|_| unreachable!())); @@ -307,10 +305,10 @@ fn a_freed_port_block_is_reused_while_its_address_is_held() { const PORTS_PER_BLOCK: usize = 256; const PORTS_PER_ADDRESS: usize = 65536 - 1024; - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let mut held = Vec::with_capacity(PORTS_PER_ADDRESS); @@ -365,10 +363,10 @@ fn the_offset_mapping_refuses_an_address_it_cannot_index() { fn an_address_past_the_indexable_span_is_refused_rather_than_panicking() { let start = u128::from(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 0)); // Far wider than the bitmap can index. - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(start, start + (1u128 << 40))], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(start, start + (1u128 << 40))], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let port = NatPort::new_port_checked(4096).unwrap_or_else(|_| unreachable!()); @@ -394,10 +392,10 @@ fn an_address_past_the_indexable_span_is_refused_rather_than_panicking() { fn ipv6_pools_allocate_within_their_range() { let start = u128::from(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 0)); let end = start + 3; - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(start, end)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(start, end)], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let mut held = Vec::new(); @@ -425,10 +423,10 @@ fn ipv6_pools_allocate_within_their_range() { #[test] fn a_live_tuple_cannot_be_reserved_again() { - let specs = vec![PoolSpec { - public_ranges: vec![AddrInterval::new(BASE, BASE)], - idle_timeout: IDLE_TIMEOUT, - }]; + let specs = vec![PoolSpec::new( + vec![AddrInterval::new(BASE, BASE)], + IDLE_TIMEOUT, + )]; let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); let held = pool_sets[0].allocate(false).expect("pool has room"); @@ -438,3 +436,39 @@ fn a_live_tuple_cannot_be_reserved_again() { drop(held); assert!(pool_sets[0].reserve(tuple.0, tuple.1).is_ok()); } + +/// What port forwarding may serve over a shared region is off limits to every owner of it, not only +/// to the expose the rule was written on: they allocate from one allocator, so a tuple one of them +/// hands out is a tuple the other cannot serve either. +#[test] +fn a_shared_region_honours_the_claims_of_every_owner() { + let address = Ipv4Addr::from(u32::try_from(BASE).unwrap_or_else(|_| unreachable!())); + let claimed = PrefixPortsSet::from([PrefixWithOptionalPorts::new( + format!("{address}/32").as_str().into(), + Some(PortRange::new(8080, 8080).unwrap_or_else(|_| unreachable!())), + )]); + // Both exposes masquerade onto the one address, so they share its region; only the first carries + // the forwarding rule. + let specs = vec![ + PoolSpec::new(vec![AddrInterval::new(BASE, BASE)], IDLE_TIMEOUT).claiming(claimed), + PoolSpec::new(vec![AddrInterval::new(BASE, BASE)], IDLE_TIMEOUT), + ]; + let pool_sets = pool_sets_for_specs::(&specs, NextHeader::TCP, false); + + let claimed_port = NatPort::new_port_checked(8080).unwrap_or_else(|_| unreachable!()); + for (owner, pool_set) in pool_sets.iter().enumerate() { + assert!( + matches!( + pool_set.reserve(address, claimed_port), + Err(AllocatorError::Denied) + ), + "owner {owner} was offered a claimed tuple" + ); + } + + // Every other port of the shared address is still theirs to hand out. + let free_port = NatPort::new_port_checked(8081).unwrap_or_else(|_| unreachable!()); + pool_sets[1] + .reserve(address, free_port) + .expect("an unclaimed port of a shared region is still available"); +} diff --git a/nat/src/masquerade/apalloc/port_alloc.rs b/nat/src/masquerade/apalloc/port_alloc.rs index 1e7661f679..c951899dc1 100644 --- a/nat/src/masquerade/apalloc/port_alloc.rs +++ b/nat/src/masquerade/apalloc/port_alloc.rs @@ -10,6 +10,7 @@ use super::NatIpWithBitmap; use super::alloc::AllocatedIp; +use super::reserved::ReservedForAddr; use crate::masquerade::allocation::AllocatorError; use crate::port::NatPort; use concurrency::concurrency_mode; @@ -87,6 +88,7 @@ pub(crate) struct PortAllocator { current_alloc_index: AtomicUsize, thread_blocks: ThreadPortMap, allocated_blocks: AllocatedPortBlockMap, + reserved: ReservedForAddr, exclude_wellknown_ports: bool, } @@ -94,11 +96,12 @@ pub(crate) struct PortAllocator { /// allocated by masquerade NAT for TCP or UDP. pub(super) const IANA_WELLKNOWN_PORT_LIMIT: u16 = 1024; -/// Number of 256-port blocks covering the IANA well-known port range (0-1023). -const IANA_WELLKNOWN_BLOCKS: u16 = IANA_WELLKNOWN_PORT_LIMIT / 256; - impl PortAllocator { - pub(crate) fn new(randomize: bool, exclude_wellknown_ports: bool) -> Self { + pub(crate) fn new( + reserved: ReservedForAddr, + randomize: bool, + exclude_wellknown_ports: bool, + ) -> Self { let mut base_ports = (0..=255).collect::>(); // Shuffle the list of port blocks for the port allocator. This way, we can pick blocks in a @@ -108,35 +111,47 @@ impl PortAllocator { if randomize { Self::shuffle_slice(&mut base_ports); } + // Count the blocks left usable rather than deriving the count, so a block excluded by both + // policies below is not subtracted twice. + let mut usable_blocks = 0u16; let blocks = std::array::from_fn(|i| { let block = AllocatorPortBlock::new(base_ports[i]); + let base_port = block.to_port_number(); // Pre-mark IANA well-known port blocks (0-1023) as permanently non-free so they are - // never handed out by masquerade NAT for TCP or UDP. - if exclude_wellknown_ports && block.to_port_number() < IANA_WELLKNOWN_PORT_LIMIT { + // never handed out by masquerade NAT for TCP or UDP. A block whose every port is claimed + // by port forwarding is likewise never handed out; a partially claimed one stays usable + // and has the claimed ports pre-set when it is allocated. + let wellknown = exclude_wellknown_ports && base_port < IANA_WELLKNOWN_PORT_LIMIT; + let claimed_whole_block = + Bitmap256::for_block(base_port, reserved.ranges(), false).bitmap_full(); + if wellknown || claimed_whole_block { block .free .store(false, concurrency::sync::atomic::Ordering::Relaxed); + } else { + usable_blocks += 1; } block }); - let usable_blocks = if exclude_wellknown_ports { - 256 - IANA_WELLKNOWN_BLOCKS - } else { - 256 - }; Self { blocks, usable_blocks: AtomicU16::new(usable_blocks), current_alloc_index: AtomicUsize::new(0), thread_blocks: ThreadPortMap::new(), allocated_blocks: AllocatedPortBlockMap::new(), + reserved, exclude_wellknown_ports, } } + /// The ports of this address that must never be handed out. + pub(crate) fn reserved_ports(&self) -> &ReservedForAddr { + &self.reserved + } + #[cfg(test)] pub(crate) fn new_no_randomness(exclude_wellknown_ports: bool) -> Self { - Self::new(false, exclude_wellknown_ports) + Self::new(ReservedForAddr::default(), false, exclude_wellknown_ports) } #[concurrency_mode(std)] @@ -238,7 +253,12 @@ impl PortAllocator { self.usable_blocks .fetch_sub(1, concurrency::sync::atomic::Ordering::Relaxed); - AllocatedPortBlock::new(ip, index, base_port_index, allow_null) + Ok(AllocatedPortBlock::new( + ip, + index, + base_port_index, + allow_null, + )) } pub(crate) fn allocate_port( @@ -294,7 +314,7 @@ impl PortAllocator { index: usize, port: NatPort, allow_null: bool, - ) -> Result>, AllocatorError> { + ) -> Arc> { self.usable_blocks .fetch_sub(1, concurrency::sync::atomic::Ordering::Relaxed); let block = Arc::new(AllocatedPortBlock::new( @@ -302,10 +322,10 @@ impl PortAllocator { index, (port.as_u16() / 256) * 256, // port block base index, discard offset within block allow_null, - )?); + )); self.allocated_blocks .insert(block.index, Arc::downgrade(&block)); - Ok(block) + block } fn find_block_for_port( @@ -317,7 +337,7 @@ impl PortAllocator { for _ in 0..BLOCK_LOOKUP_ATTEMPTS { let (block_was_free, index) = self.try_to_reserve_block(port)?; if block_was_free { - return self.allocate_block_for_reservation(ip, index, port, allow_null); + return Ok(self.allocate_block_for_reservation(ip, index, port, allow_null)); } if let Some(block) = self.allocated_blocks.search_for_block(port) { return Ok(block); @@ -342,6 +362,13 @@ impl PortAllocator { debug!("Explicit reservation for well-known port {port} denied by allocator policy"); return Err(AllocatorError::Denied); } + // Same reason, for the ports port forwarding claims. Answering here matters beyond the error + // being clearer: a block claimed in full is never allocated, so `find_block_for_port` would + // find it neither free nor in the allocated map, and spend every retry before giving up. + if self.reserved.contains(port.as_u16()) { + debug!("Explicit reservation for port-forwarding port {port} denied"); + return Err(AllocatorError::Denied); + } let block = self.find_block_for_port(ip, port)?; block.reserve_port_from_block(port) } @@ -374,25 +401,17 @@ pub(crate) struct AllocatedPortBlock { } impl AllocatedPortBlock { - fn new( - ip: Arc>, - index: usize, - base_port_idx: u16, - allow_null: bool, - ) -> Result { - let block = Self { + fn new(ip: Arc>, index: usize, base_port_idx: u16, allow_null: bool) -> Self { + // The claims are taken from the address rather than passed in, so that every path creating a + // block honours the ports port forwarding may use, whether it allocates or reserves. + let usage_bitmap = + Bitmap256::for_block(base_port_idx, ip.reserved_ports().ranges(), !allow_null); + Self { ip, base_port_idx, index, - usage_bitmap: Mutex::new(Bitmap256::new()), - }; - if !allow_null && block.base_port_idx == 0 { - let mut mutex_guard = block.usage_bitmap.lock(); - mutex_guard.reserve_port_from_bitmap(0).map_err(|()| { - AllocatorError::InternalIssue("Failed to reserve port 0 from new block".to_string()) - })?; + usage_bitmap: Mutex::new(usage_bitmap), } - Ok(block) } fn ip(&self) -> I { @@ -740,6 +759,57 @@ impl Bitmap256 { } } + /// The starting state of the block based at `base_port`: every port `claimed` covers is marked + /// used before the block is handed out, so it can never be allocated. Port 0 is treated the same + /// way when it may not be given out. + /// + /// A claim is clipped to this block, so one spanning several blocks marks the right ports in each + /// of them. [`Self::bitmap_full`] then says whether the block has anything left to allocate. + fn for_block(base_port: u16, claimed: &[PortRange], reserve_null: bool) -> Self { + debug_assert_eq!(base_port % 256, 0, "not a port block base: {base_port}"); + let mut bitmap = Self::new(); + if reserve_null && base_port == 0 { + bitmap.first_half |= 1; + } + for range in claimed { + let start = range.start().max(base_port); + let end = range.end().min(base_port | 0xff); + if start > end { + continue; + } + // Both offsets fall inside the block, so they fit in a u8. + let offset = |port: u16| { + u8::try_from(port - base_port).unwrap_or_else(|_| unreachable!("{port}")) + }; + bitmap.reserve_offset_range(offset(start), offset(end)); + } + bitmap + } + + /// Mark the inclusive offset range `start..=end` of the block used. + fn reserve_offset_range(&mut self, start: u8, end: u8) { + debug_assert!(start <= end, "start: {start}, end: {end}"); + if start < 128 { + self.first_half |= Self::contiguous_bits(start, end.min(127)); + } + if end >= 128 { + self.second_half |= Self::contiguous_bits(start.max(128) - 128, end - 128); + } + } + + /// Ones in `start..=end`, both offsets within one half. + fn contiguous_bits(start: u8, end: u8) -> u128 { + debug_assert!(start <= end && end < 128, "start: {start}, end: {end}"); + let width = u32::from(end - start) + 1; + // A full half cannot be built by shifting: `1 << 128` overflows. + let ones = if width >= 128 { + u128::MAX + } else { + (1u128 << width) - 1 + }; + ones << start + } + fn bitmap_full(&self) -> bool { self.first_half == u128::MAX && self.second_half == u128::MAX } @@ -881,6 +951,7 @@ fn collect_ranges_from_u128_bitmap(bitmap: u128, base: u16) -> BTreeSet Vec { + pairs + .iter() + .map(|&(start, end)| PortRange::new(start, end).unwrap()) + .collect() + } + + #[test] + fn a_claim_outside_the_block_marks_nothing() { + let bitmap = Bitmap256::for_block(1024, &claimed(&[(300, 400)]), false); + assert!((0..=255u8).all(|port| !port_is_used(&bitmap, port))); + } + + #[test] + fn a_claim_is_clipped_to_the_block() { + let bitmap = Bitmap256::for_block(1024, &claimed(&[(1000, 1100)]), false); + assert!(!bitmap.bitmap_full()); + assert!(port_is_used(&bitmap, 0), "port 1024 must be reserved"); + assert!(port_is_used(&bitmap, 76), "port 1100 must be reserved"); + assert!(!port_is_used(&bitmap, 77), "port 1101 must be free"); + } + + #[test] + fn a_claim_spanning_both_halves_marks_both() { + // Ports 1100-1200 are offsets 76-176 of the block based at 1024, crossing the boundary + // between the two halves the bitmap is stored in. + let bitmap = Bitmap256::for_block(1024, &claimed(&[(1100, 1200)]), false); + assert!(port_is_used(&bitmap, 127), "port 1151 must be reserved"); + assert!(port_is_used(&bitmap, 128), "port 1152 must be reserved"); + assert!(!port_is_used(&bitmap, 177), "port 1201 must be free"); + assert!(!port_is_used(&bitmap, 255), "port 1279 must be free"); + } + + #[test] + fn a_claim_running_past_the_block_marks_its_tail() { + let bitmap = Bitmap256::for_block(1024, &claimed(&[(1100, 1300)]), false); + assert!(port_is_used(&bitmap, 76), "port 1100 must be reserved"); + assert!(port_is_used(&bitmap, 255), "port 1279 must be reserved"); + assert!(!port_is_used(&bitmap, 75), "port 1099 must be free"); + } + + // A block with nothing left to allocate, which is how `PortAllocator::new` recognises the blocks + // it must never hand out. + #[test] + fn a_claim_over_a_whole_block_fills_it() { + assert!( + Bitmap256::for_block(1024, &claimed(&[(1024, 1279)]), false).bitmap_full(), + "an exact claim must fill the block" + ); + // And so must a claim spanning it, without leaking into its neighbours. + let wide = claimed(&[(1000, 2000)]); + assert!(Bitmap256::for_block(1024, &wide, false).bitmap_full()); + assert!(Bitmap256::for_block(1280, &wide, false).bitmap_full()); + assert!(!Bitmap256::for_block(2048, &wide, false).bitmap_full()); + } + + #[test] + fn disjoint_claims_accumulate_in_one_block() { + let bitmap = Bitmap256::for_block(1024, &claimed(&[(1024, 1024), (1279, 1279)]), false); + assert!(port_is_used(&bitmap, 0)); + assert!(port_is_used(&bitmap, 255)); + assert!(!port_is_used(&bitmap, 128)); + } + + #[test] + fn port_zero_is_reserved_in_the_first_block_alone() { + assert!(port_is_used(&Bitmap256::for_block(0, &[], true), 0)); + assert!(!port_is_used(&Bitmap256::for_block(0, &[], false), 0)); + // Offset 0 of any other block is a legitimate port. + assert!(!port_is_used(&Bitmap256::for_block(256, &[], true), 0)); + } + + /// A block base and a handful of claims around it. + #[derive(Debug, Clone)] + struct Claims { + base_port: u16, + ranges: Vec<(u16, u16)>, + } + + impl TypeGenerator for Claims { + fn generate(driver: &mut D) -> Option { + let base_port = u16::from(driver.produce::()?) * 256; + let count = usize::from(driver.produce::()? % 4); + let mut ranges = Vec::with_capacity(count); + for _ in 0..count { + // Draw around the block so claims land inside it, across its edges, and outside. + let start = base_port.saturating_sub(300).saturating_add( + u16::from(driver.produce::()?) * 4 + u16::from(driver.produce::()? % 4), + ); + let end = start.saturating_add(u16::from(driver.produce::()?) * 3); + ranges.push((start, end)); + } + Some(Self { base_port, ranges }) + } + } + + #[test] + fn a_block_holds_exactly_the_claimed_ports_of_its_block() { + bolero::check!() + .with_type() + .cloned() + .for_each(|claims: Claims| { + let ranges = claimed(&claims.ranges); + let bitmap = Bitmap256::for_block(claims.base_port, &ranges, false); + // The claimed set, restricted to this block, port by port. + let is_claimed = |port: u16| { + ranges + .iter() + .any(|range| range.start() <= port && port <= range.end()) + }; + + for offset in 0..=255u8 { + let port = claims.base_port + u16::from(offset); + assert_eq!( + port_is_used(&bitmap, offset), + is_claimed(port), + "port {port} (offset {offset} of block {})", + claims.base_port + ); + } + assert_eq!( + bitmap.bitmap_full(), + (0..=255u8).all(|offset| is_claimed(claims.base_port + u16::from(offset))) + ); + }); + } } diff --git a/nat/src/masquerade/apalloc/reserved.rs b/nat/src/masquerade/apalloc/reserved.rs new file mode 100644 index 0000000000..3507ce9a5e --- /dev/null +++ b/nat/src/masquerade/apalloc/reserved.rs @@ -0,0 +1,220 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +//! Adds types for IP/port tuples that masquerade may not hand out since they may be +//! reserved by port forwarding. +//! +//! Port forwarding maps a public address and port onto a private endpoint, and a masquerade expose +//! may cover that same public address. Both would then claim one public tuple, and since a +//! [`FlowKey`](net::FlowKey) carries no notion of which translation owns it, the resulting flows can +//! collide: an inbound packet for the forwarded service is instead matched by a masquerade flow and +//! delivered to an unrelated private endpoint. +//! +//! Which tuples port forwarding may claim follows from the configuration, so they are removed from +//! the masquerade pools while those pools are built rather than defended per flow. The reserved bits +//! are then indistinguishable from allocated ones, and no allocation path has to know that port +//! forwarding exists. + +use lpm::prefix::{PortRange, Prefix, PrefixPortsSet}; +use std::net::IpAddr; + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +/// Ports of the (public) address space that port forwarding may claim, as (prefix, ports) pairs. +pub(crate) struct ReservedPorts(Vec<(Prefix, PortRange)>); +impl ReservedPorts { + /// Build a `ReservedPorts` from a `PrefixPortsSet`. The object contains the set of (prefix, ports) + /// that may be reserved upfront and not handed out. + /// + /// Every pair is kept: two rules may claim different ports of one address, and collapsing them + /// to a single range per address would lose one of the claims. + /// + /// Prefixes carrying no port range are skipped since port-forwarding always specifies port-ranges. + pub(crate) fn from_set(set: &PrefixPortsSet) -> Self { + Self( + set.iter() + .filter_map(|prefix| Some((prefix.prefix(), prefix.ports()?))) + .collect(), + ) + } + + /// The port ranges reserved for a given IP address, computed as the union of the + /// ports reserved across all prefixes that contain this address. Since the ports + /// reserved for several prefixes may overlap, the ports reserved for an address + /// are merged here, leaving the smallest set of disjoint, non-adjancent ranges. + #[must_use] + pub(crate) fn for_addr(&self, addr: IpAddr) -> ReservedForAddr { + // collect all the (prefix, port-range) that include (cover) the given address + let mut covering_ranges: Vec = self + .0 + .iter() + .filter(|(prefix, _)| prefix.covers_addr(&addr)) + .map(|(_, ports)| *ports) + .collect(); + + // Sort so that we can merge and accumulate in one pass + covering_ranges.sort_unstable(); + + let mut merged: Vec = Vec::with_capacity(covering_ranges.len()); + for range in covering_ranges { + match merged.pop() { + Some(last) => match last.merge(range) { + Some(union) => merged.push(union), + None => merged.extend([last, range]), + }, + None => merged.push(range), + } + } + + // build + ReservedForAddr(merged) + } +} + +/// The reserved port ranges that apply to one (public) address. +/// +/// Empty for most addresses, and never large: it holds the disjoint ranges that the claims covering +/// that address merge into, so at most one range per configured claim. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(crate) struct ReservedForAddr(Vec); + +impl ReservedForAddr { + #[must_use] + #[cfg(test)] + pub(crate) fn is_empty(&self) -> bool { + self.0.is_empty() + } + + /// The reserved ranges, for whoever has to keep those ports out of a bitmap. + #[must_use] + pub(crate) fn ranges(&self) -> &[PortRange] { + &self.0 + } + + /// Whether a specific port is reserved. + #[must_use] + pub(crate) fn contains(&self, port: u16) -> bool { + self.0 + .iter() + .any(|range| range.start() <= port && port <= range.end()) + } + + /// Tell the number of ports reserved. We can sum the ports of each range since + /// all of the ranges are disjoint by construction. + #[must_use] + #[cfg(test)] + pub(crate) fn reserved(&self) -> usize { + self.0.iter().map(PortRange::len).sum() + } +} + +#[cfg(test)] +mod tests { + use super::{ReservedForAddr, ReservedPorts}; + use lpm::prefix::{PortRange, PrefixPortsSet, PrefixWithOptionalPorts}; + use std::net::IpAddr; + + fn claim(prefix: &str, start: u16, end: u16) -> PrefixWithOptionalPorts { + PrefixWithOptionalPorts::new(prefix.into(), Some(PortRange::new(start, end).unwrap())) + } + + fn addr(addr: &str) -> IpAddr { + addr.parse().unwrap() + } + + fn ports(reserved: &ReservedForAddr) -> Vec<(u16, u16)> { + reserved + .ranges() + .iter() + .map(|range| (range.start(), range.end())) + .collect() + } + + #[test] + fn claims_on_one_prefix_all_survive() { + // The defect the previous static implementation had: a single-range-per-address model kept + // only one of these two claims. + let reserved = ReservedPorts::from_set(&PrefixPortsSet::from([ + claim("5.6.7.8/32", 8080, 8080), + claim("5.6.7.8/32", 9090, 9090), + ])); + + let for_addr = reserved.for_addr(addr("5.6.7.8")); + assert_eq!(ports(&for_addr), [(8080, 8080), (9090, 9090)]); + assert!(for_addr.contains(8080), "first claim was lost"); + assert!(for_addr.contains(9090), "second claim was lost"); + assert!(!for_addr.contains(8081)); + } + + #[test] + fn overlapping_and_adjacent_claims_merge() { + let reserved = ReservedPorts::from_set(&PrefixPortsSet::from([ + claim("5.6.7.8/32", 100, 200), // overlaps the claim below + claim("5.6.7.8/32", 150, 300), + claim("5.6.7.8/32", 301, 400), // abuts the claim above: no port between the two + claim("5.6.7.8/32", 500, 600), + claim("5.6.7.8/32", 520, 540), // contained in the claim above + ])); + + // The first three claims describe one uninterrupted run of ports, and the last two another. + let for_addr = reserved.for_addr(addr("5.6.7.8")); + assert_eq!(ports(&for_addr), [(100, 400), (500, 600)]); + assert!(for_addr.contains(400)); + assert!(!for_addr.contains(401), "the gap between the runs was lost"); + assert!(for_addr.contains(540)); + } + + #[test] + fn claims_apply_only_to_the_addresses_they_cover() { + let reserved = + ReservedPorts::from_set(&PrefixPortsSet::from([claim("5.6.7.0/30", 4000, 4001)])); + + assert!(reserved.for_addr(addr("5.6.7.1")).contains(4000)); + assert!(reserved.for_addr(addr("5.6.7.4")).is_empty()); + // An address of the other version is never covered. + assert!(reserved.for_addr(addr("::1")).is_empty()); + } + + #[test] + fn a_prefix_without_ports_claims_nothing() { + let set = PrefixPortsSet::from([PrefixWithOptionalPorts::new("5.6.7.8/32".into(), None)]); + assert!(ReservedPorts::from_set(&set).0.is_empty()); + } + + #[test] + fn test_combined_reservations() { + let claim1 = claim("100.64.1.0/25", 15, 100); + let claim2 = claim("100.64.1.128/25", 20, 60); + let claim3 = claim("100.64.1.192/26", 10, 180); + + let all_claims = PrefixPortsSet::from([claim1, claim2, claim3]); + let reserved = ReservedPorts::from_set(&all_claims); + + let address = addr("100.64.1.1"); + let res = reserved.for_addr(address); + assert_eq!(res.reserved(), 100 - 15 + 1); + + assert!(!res.contains(14)); + assert!(res.contains(15)); + assert!(res.contains(100)); + assert!(!res.contains(101)); + + let address = addr("100.64.1.129"); + let res = reserved.for_addr(address); + assert_eq!(res.reserved(), 60 - 20 + 1); + + assert!(!res.contains(19)); + assert!(res.contains(20)); + assert!(res.contains(60)); + assert!(!res.contains(61)); + + let address = addr("100.64.1.193"); + let res = reserved.for_addr(address); + assert_eq!(res.reserved(), 180 - 10 + 1); + + assert_eq!(ports(&res), [(10, 180)]); + assert!(!res.contains(9)); + assert!(res.contains(10)); + assert!(res.contains(180)); + assert!(!res.contains(181)); + } +} diff --git a/nat/src/masquerade/apalloc/setup.rs b/nat/src/masquerade/apalloc/setup.rs index 8465b7ae4e..c6f018ee23 100644 --- a/nat/src/masquerade/apalloc/setup.rs +++ b/nat/src/masquerade/apalloc/setup.rs @@ -3,39 +3,43 @@ //! Build masquerade pools by grouping exposes per peer VPC and splitting overlapping public ranges //! into disjoint, shared regions. +//! +//! This is also where the public tuples port forwarding may claim are withheld from the pools, so +//! that the two translations of a peering cannot end up claiming one public tuple. use super::alloc::{IpAllocator, NatPool, PoolSet}; use super::region::{AddrInterval, Region, decompose, regions_by_owner}; +use super::reserved::ReservedPorts; use super::{NatAllocator, NatIpWithBitmap, PoolTable, PoolTableKey}; use crate::masquerade::allocator_writer::MasqueradeConfig; use crate::masquerade::natip::NatIp; use config::external::overlay::vpcpeering::{ValidatedExpose, ValidatedManifest}; -use lpm::prefix::{PrefixPortsSet, PrefixWithOptionalPorts}; +use lpm::prefix::{L4Protocol, PrefixPortsSet, PrefixWithOptionalPorts}; use net::ip::NextHeader; use net::packet::VpcDiscriminant; use std::collections::BTreeMap; +use std::net::{Ipv4Addr, Ipv6Addr}; use std::time::Duration; -use tracing::debug; +use tracing::{debug, error}; const DEFAULT_MASQUERADE_IDLE_TIMEOUT: Duration = Duration::from_mins(2); impl NatAllocator { - pub(crate) fn build_pools(&mut self, config: &MasqueradeConfig) { + pub(crate) fn build_pool44(config: &MasqueradeConfig) -> PoolTable { build_pools_generic( config, ValidatedManifest::masquerade_exposes_44, - &mut self.pools_src44, + ValidatedManifest::port_forwarding_exposes_44, NextHeader::ICMP, - self.randomize, - ); - + ) + } + pub(crate) fn build_pool66(config: &MasqueradeConfig) -> PoolTable { build_pools_generic( config, ValidatedManifest::masquerade_exposes_66, - &mut self.pools_src66, + ValidatedManifest::port_forwarding_exposes_66, NextHeader::ICMP6, - self.randomize, - ); + ) } } @@ -51,22 +55,62 @@ struct GatheredExpose<'a> { // The public range this expose allocates from, as raw address intervals. public_ranges: Vec, idle_timeout: Duration, + // The port-forwarding exposes of the same peering, whose public tuples this expose's pools must + // leave alone. + port_forwarding: Vec<&'a ValidatedExpose>, +} + +/// The public tuples the given port-forwarding exposes may claim for one protocol. +/// +/// A claim bears only on the protocol its rule forwards, so a TCP rule leaves the UDP pools alone. +/// ICMP pools are never claimed from: port forwarding is validated to carry port ranges, which +/// identifiers are not. +/// +/// The claims are deliberately not intersected with the masquerade ranges: a claim on an address no +/// masquerade pool holds costs nothing, because the pools only ever consult the claims covering the +/// addresses they own. That leaves one description of which tuples port forwarding may use, rather +/// than a second one here that has to keep agreeing with the port forwarder. +fn claims_for(protocol: NextHeader, exposes: &[&ValidatedExpose]) -> PrefixPortsSet { + let wanted = match protocol { + NextHeader::TCP => L4Protocol::Tcp, + NextHeader::UDP => L4Protocol::Udp, + _ => return PrefixPortsSet::new(), + }; + + let mut claimed = PrefixPortsSet::new(); + for expose in exposes { + // A port-forwarding expose is validated to carry NAT configuration. + let Some(nat) = expose.nat() else { + error!("Port-forwarding expose without NAT configuration. This is a bug"); + continue; + }; + if nat.proto.intersection(&wanted).is_some() { + claimed.extend(expose.as_range_or_empty().clone()); + } + } + claimed } // Exposes toward different peers may safely reuse the same public range. -fn gather_exposes<'a, J, F, FIter>( +fn gather_exposes<'a, J, F, FIter, P, PIter>( config: &'a MasqueradeConfig, exposes_filter: &F, + port_forwarding_filter: &P, ) -> BTreeMap>> where J: NatIp, F: Fn(&'a ValidatedManifest) -> FIter, FIter: Iterator, + P: Fn(&'a ValidatedManifest) -> PIter, + PIter: Iterator, { let mut groups: BTreeMap>> = BTreeMap::new(); for nat_peering in config.iter() { let manifest = nat_peering.peering.local(); + // Port forwarding and masquerade of one peering share a public space, so the exposes of this + // manifest are the ones whose claims the pools built from it must honour. + let port_forwarding: Vec<&'a ValidatedExpose> = port_forwarding_filter(manifest).collect(); for expose in exposes_filter(manifest) { let public_ranges = public_intervals::(expose.as_range_or_empty()); if public_ranges.is_empty() { @@ -84,6 +128,7 @@ where idle_timeout: expose .idle_timeout() .unwrap_or(DEFAULT_MASQUERADE_IDLE_TIMEOUT), + port_forwarding: port_forwarding.clone(), }); } } @@ -110,19 +155,23 @@ fn public_intervals(ranges: &PrefixPortsSet) -> Vec { // Building /////////////////////////////////////////////////////////////////////////////// -fn build_pools_generic<'a, I, J, F, FIter>( +fn build_pools_generic<'a, I, J, F, FIter, P, PIter>( config: &'a MasqueradeConfig, exposes_filter: F, - table: &mut PoolTable, + port_forwarding_filter: P, icmp_proto: NextHeader, - randomize: bool, -) where +) -> PoolTable +where I: NatIpWithBitmap, J: NatIpWithBitmap, F: Fn(&'a ValidatedManifest) -> FIter, FIter: Iterator, + P: Fn(&'a ValidatedManifest) -> PIter, + PIter: Iterator, { - let groups = gather_exposes::(config, &exposes_filter); + let mut table = PoolTable::new(); + let randomize = config.randomize(); + let groups = gather_exposes::(config, &exposes_filter, &port_forwarding_filter); for (dst_vpc_id, exposes) in groups { // Allocations for TCP, for example, do not affect allocations for UDP or for ICMP: the @@ -133,6 +182,7 @@ fn build_pools_generic<'a, I, J, F, FIter>( .iter() .map(|expose| PoolSpec { public_ranges: expose.public_ranges.clone(), + claimed: claims_for(protocol, &expose.port_forwarding), idle_timeout: expose.idle_timeout, }) .collect(); @@ -140,7 +190,7 @@ fn build_pools_generic<'a, I, J, F, FIter>( let pool_sets = pool_sets_for_specs::(&specs, protocol, randomize); for (expose, pool_set) in exposes.iter().zip(pool_sets) { add_pool_entries( - table, + &mut table, expose.private_prefixes, expose.src_vpc_id, dst_vpc_id, @@ -150,15 +200,39 @@ fn build_pools_generic<'a, I, J, F, FIter>( } } } + table } /// The config-independent inputs for one expose's pools. -#[derive(Clone)] +#[derive(Clone, Default)] pub(crate) struct PoolSpec { pub(crate) public_ranges: Vec, + /// The public tuples port forwarding may claim over this expose's ranges, for the protocol the + /// pools are being built for. + pub(crate) claimed: PrefixPortsSet, pub(crate) idle_timeout: Duration, } +impl PoolSpec { + /// A spec with nothing claimed by port forwarding. + #[cfg(test)] + pub(crate) fn new(public_ranges: Vec, idle_timeout: Duration) -> Self { + Self { + public_ranges, + idle_timeout, + ..Self::default() + } + } + + /// The same spec, with public tuples a forwarding rule may serve. + #[cfg(test)] + #[must_use] + pub(crate) fn claiming(mut self, claimed: PrefixPortsSet) -> Self { + self.claimed = claimed; + self + } +} + /// Cut the space the given exposes claim into disjoint regions, build one allocator per region, /// and return the pools each expose may allocate from, in the same order as `specs`. /// @@ -181,7 +255,7 @@ pub(crate) fn pool_sets_for_specs( specs.len() ); - let allocators = build_region_allocators::(®ions, protocol, randomize); + let allocators = build_region_allocators::(®ions, specs, protocol, randomize); let by_owner = regions_by_owner(®ions); specs @@ -204,6 +278,7 @@ pub(crate) fn pool_sets_for_specs( // keeps a public address and port from being handed out twice. fn build_region_allocators( regions: &[Region], + specs: &[PoolSpec], protocol: NextHeader, randomize: bool, ) -> Vec> { @@ -214,7 +289,20 @@ fn build_region_allocators( regions .iter() .map(|region| { - let pool = NatPool::for_range(region.range, exclude_wellknown_ports); + // A region is shared, so it must honour every claim on it: what port forwarding may take + // from any of its owners is off limits to all of them. + let claimed = region + .owners + .iter() + .fold(PrefixPortsSet::new(), |accumulated, &owner| { + accumulated.union_prefixes_and_ports(&specs[owner].claimed) + }); + + let pool = NatPool::for_range( + region.range, + ReservedPorts::from_set(&claimed), + exclude_wellknown_ports, + ); IpAllocator::new(pool, randomize) }) .collect() diff --git a/nat/src/masquerade/apalloc/test_alloc.rs b/nat/src/masquerade/apalloc/test_alloc.rs index 100a9bfc44..ecd718dbd3 100644 --- a/nat/src/masquerade/apalloc/test_alloc.rs +++ b/nat/src/masquerade/apalloc/test_alloc.rs @@ -13,6 +13,7 @@ mod context { use crate::masquerade::apalloc::{NatAllocator, PoolTable, PoolTableKey}; use config::external::overlay::vpc::{Peering, ValidatedVpcTable, Vpc, VpcTable}; use config::external::overlay::vpcpeering::{VpcExpose, VpcManifest}; + use lpm::prefix::{L4Protocol, PortRange, PrefixWithOptionalPorts}; use net::ip::NextHeader; use net::packet::VpcDiscriminant; use net::udp::UdpPort; @@ -157,6 +158,55 @@ mod context { NatAllocator::new(config, 1) } + #[allow(dead_code)] + fn prefix_with_ports(prefix: &str, start: u16, end: u16) -> PrefixWithOptionalPorts { + PrefixWithOptionalPorts::new(prefix.into(), Some(PortRange::new(start, end).unwrap())) + } + + // Masquerade onto one public address, with a forwarding rule on port 8080 of it for one protocol + // only. VPC-1 masquerades towards VPC-2. + #[allow(dead_code)] + fn build_context_port_forward_for(protocol: L4Protocol) -> ValidatedVpcTable { + let masquerade = VpcExpose::empty() + .make_masquerade(None) + .unwrap() + .ip("1.1.0.0/16".into()) + .as_range("10.1.0.0/32".into()) + .unwrap(); + let forward = VpcExpose::empty() + .make_port_forwarding(None, Some(protocol)) + .unwrap() + .ip(prefix_with_ports("1.1.0.5/32", 8080, 8080)) + .as_range(prefix_with_ports("10.1.0.0/32", 8080, 8080)) + .unwrap(); + let local = VpcManifest::with_exposes("VPC-1", vec![masquerade, forward]); + let remote = + VpcManifest::with_exposes("VPC-2", vec![VpcExpose::empty().ip("3.0.0.0/24".into())]); + + let mut vpc1 = Vpc::new("VPC-1", "67890", vni1().as_u32()).unwrap(); + let vpc2 = Vpc::new("VPC-2", "12345", vni2().as_u32()).unwrap(); + vpc1.peerings.push(Peering { + name: "port_forward_peering".into(), + local, + remote, + remote_id: "12345".try_into().unwrap(), + remote_vni: vpc2.vni, + gwgroup: "default".into(), + acl: None, + }); + + let mut vpctable = VpcTable::new(); + vpctable.add(vpc1).unwrap(); + vpctable.add(vpc2).unwrap(); + vpctable.validate().unwrap() + } + + #[allow(dead_code)] + pub fn build_allocator_port_forward_for(protocol: L4Protocol) -> NatAllocator { + let vpc_table = build_context_port_forward_for(protocol); + NatAllocator::new(MasqueradeConfig::new(&vpc_table).set_randomize(false), 1) + } + // Two VPCs masquerade onto one public range toward the same peer. #[allow(dead_code)] fn build_context_shared_public_range() -> ValidatedVpcTable { @@ -397,13 +447,9 @@ mod context { mod tests { use super::context::*; - use crate::NatPort; - use crate::masquerade::allocation::AllocatorError; use concurrency::sync::Arc; use concurrency::thread; use net::ip::NextHeader; - use std::net::IpAddr; - use std::num::NonZero; #[allow(dead_code)] pub(super) fn concurrent_allocations() { @@ -450,75 +496,21 @@ mod tests { assert_eq!(bitmap.len(), 3); // 3 IP addresses available to NAT 1.1.0.0 assert!(in_use.front().unwrap().upgrade().is_none()); // Weak references in list no longer resolve } - - #[test] - fn port_forward_flows_share_and_release_a_public_tuple() { - let allocator = build_allocator(); - let public_ip = IpAddr::V4(addr_v4("10.1.0.0")); - let public_port = NonZero::new(1024).unwrap(); - - let first = allocator - .reserve_port_forward(NextHeader::TCP, vpcd2(), public_ip, public_port) - .unwrap() - .expect("the tuple overlaps a masquerade pool"); - let second = allocator - .reserve_port_forward(NextHeader::TCP, vpcd2(), public_ip, public_port) - .unwrap() - .expect("the tuple overlaps a masquerade pool"); - assert!(Arc::ptr_eq(&first, &second)); - - let port = NatPort::new_port(public_port); - match allocator.reserve_port( - NextHeader::TCP, - vpcd1(), - vpcd2(), - ipaddr("1.1.0.1"), - public_ip, - port, - ) { - Err(AllocatorError::PortReservationFailed(blocked)) => { - assert_eq!(blocked, public_port.get()); - } - other => panic!("a live port-forward lease must block the reservation, got {other:?}"), - } - - drop((first, second)); - assert!( - allocator - .reserve_port( - NextHeader::TCP, - vpcd1(), - vpcd2(), - ipaddr("1.1.0.1"), - public_ip, - port, - ) - .is_ok() - ); - } - - #[test] - fn well_known_port_forwards_need_no_lease() { - let allocator = build_allocator(); - let lease = allocator - .reserve_port_forward( - NextHeader::TCP, - vpcd2(), - ipaddr("10.1.0.0"), - NonZero::new(80).unwrap(), - ) - .unwrap(); - assert!(lease.is_none()); - } } #[concurrency_mode(std)] mod std_tests { use super::context::*; + use crate::NatPort; + use crate::masquerade::allocation::AllocatorError; use crate::masquerade::apalloc::PoolTableKey; - use crate::masquerade::apalloc::alloc::PoolRegion; + use crate::masquerade::apalloc::alloc::{IpAllocator, NatPool, PoolRegion}; + use crate::masquerade::apalloc::region::AddrInterval; + use crate::masquerade::apalloc::reserved::ReservedPorts; + use lpm::prefix::{L4Protocol, PortRange, PrefixPortsSet, PrefixWithOptionalPorts}; use net::ip::NextHeader; - use std::net::IpAddr; + use std::net::{IpAddr, Ipv4Addr}; + use std::num::NonZero; #[test] fn test_build_allocator() { @@ -1008,6 +1000,105 @@ mod std_tests { .count(); assert_eq!(tcp_entries, 2); } + + /////////////////////////////////////////////////////////////////////////// + // Ports claimed by port forwarding + /////////////////////////////////////////////////////////////////////////// + + // A pool over one public address, claiming the whole 1024-1279 block and the single port 1400, + // built the way `build_region_allocators` builds one. + fn allocator_with_claims() -> (IpAllocator, Ipv4Addr) { + let address = addr_v4("10.1.0.0"); + let bits = u128::from(addr_v4_bits("10.1.0.0")); + let claims = PrefixPortsSet::from([ + PrefixWithOptionalPorts::new( + "10.1.0.0/32".into(), + Some(PortRange::new(1024, 1279).unwrap()), + ), + PrefixWithOptionalPorts::new( + "10.1.0.0/32".into(), + Some(PortRange::new(1400, 1400).unwrap()), + ), + ]); + let pool = NatPool::for_range( + AddrInterval::new(bits, bits), + ReservedPorts::from_set(&claims), + true, + ); + (IpAllocator::new(pool, false), address) + } + + #[test] + fn claimed_ports_are_never_allocated() { + let (allocator, _) = allocator_with_claims(); + + // Enough to walk past the claimed block and all the way through the next one, which holds + // the single claimed port. + let held: Vec<_> = (0..512) + .map(|_| allocator.allocate(false).expect("the pool has room")) + .collect(); + let ports: Vec = held.iter().map(|port| port.port().as_u16()).collect(); + + assert!( + ports.iter().all(|&port| !(1024..=1279).contains(&port)), + "a fully claimed block was handed out" + ); + assert!(!ports.contains(&1400), "a claimed port was handed out"); + // The claimed block is skipped rather than partially used: allocation starts after it. + assert_eq!(ports[0], 1280); + // Its neighbour is used, minus the one port claimed inside it. + assert!(ports.contains(&1399) && ports.contains(&1401)); + } + + /// A forwarding rule claims tuples of the protocol it forwards. The address's other protocols + /// have their own port space, which no flow of the forwarded protocol can collide with. + #[test] + fn a_claim_is_confined_to_the_protocol_it_forwards() { + let port = NatPort::new_port(NonZero::new(8080).unwrap()); + let public = ipaddr("10.1.0.0"); + let private = ipaddr("1.1.0.1"); + + for (forwarded, claimed, untouched) in [ + (L4Protocol::Tcp, NextHeader::TCP, NextHeader::UDP), + (L4Protocol::Udp, NextHeader::UDP, NextHeader::TCP), + ] { + let allocator = build_allocator_port_forward_for(forwarded); + assert!( + matches!( + allocator.reserve_port(claimed, vpcd1(), vpcd2(), private, public, port), + Err(AllocatorError::Denied) + ), + "a {forwarded:?} rule must claim the {claimed} tuple" + ); + allocator + .reserve_port(untouched, vpcd1(), vpcd2(), private, public, port) + .unwrap_or_else(|e| { + panic!("a {forwarded:?} rule must claim nothing from {untouched}: {e}") + }); + } + } + + #[test] + fn claimed_ports_cannot_be_reserved() { + let (allocator, address) = allocator_with_claims(); + + for claimed in [1024, 1279, 1400] { + let port = NatPort::new_port(NonZero::new(claimed).unwrap()); + assert!( + matches!( + allocator.reserve(address, port), + Err(AllocatorError::Denied) + ), + "reserving claimed port {claimed} must be denied" + ); + } + + // A port outside every claim is still reservable on the same address. + let free = NatPort::new_port(NonZero::new(5000).unwrap()); + allocator + .reserve(address, free) + .expect("an unclaimed port is reservable"); + } } // Loom's Weak shim keeps allocator liveness entries alive forever. diff --git a/nat/src/masquerade/flows.rs b/nat/src/masquerade/flows.rs index 69a7408e9b..a2d5729c05 100644 --- a/nat/src/masquerade/flows.rs +++ b/nat/src/masquerade/flows.rs @@ -5,7 +5,6 @@ use crate::NatPort; use crate::common::NatAction; use crate::masquerade::apalloc::NatAllocator; use crate::masquerade::state::MasqueradeState; -use crate::portfw::update_port_forward_lease; use config::GenId; use flow_entry::flow_table::{FlowTable, FlowTableReadGuard}; @@ -16,17 +15,18 @@ use net::flows::FlowInfo; use std::net::IpAddr; use tracing::{debug, error}; -/// Detach flows from an allocator before it is removed. -pub(crate) fn remove_allocator_from_flows(flow_table: &FlowTable) { +/// Invalidate all masquerading flows +pub(crate) fn invalidate_masquerade_flows(flow_table: &FlowTable) { + debug!("INVALIDATING all masquerading flows..."); flow_table.for_each_flow(|_key, flow_info| { if flow_info.locked.read().nat_state.as_ref().is_some() { flow_info.invalidate_pair(); } - let _ = update_port_forward_lease(flow_info, None); }); + debug!("INVALIDATING all masquerading flows COMPLETED"); } -/// Upgrade to genid `GenId` all of the flows in the flow table +/// Upgrade to genid `GenId` all of the active masquerading flows pub(crate) fn upgrade_all_masquerading_flows(flow_table: &FlowTable, genid: GenId) { debug!("UPGRADING all masquerading flows to gen {genid}..."); let mut count = 0; @@ -49,6 +49,8 @@ fn get_flow_masquerading_allocation(flow_info: &FlowInfo) -> Option<(IpAddr, Nat .nat_state .extract_ref::()? .allocation()?; + + debug_assert!(flow_info.get_flags().is_initiator()); Some((alloc.ip(), alloc.port())) } @@ -94,17 +96,19 @@ pub(crate) fn check_masquerading_flow( flow_info: &FlowInfo, allocator: &NatAllocator, ) { + // Skip flows that are up-to-date (this could be done by iterator) let config = allocator.config(); let genid = allocator.genid(); if flow_info.genid() == genid { return; } - // ip and port allocated to masquerade a flow. If masquerading flow did not have allocated port - // we skip it since we will invalidate it (or upgrade it) with the related flow that has an allocation. + // get ip + port allocated to flow. If flow does not have allocated port, skip it since we will + // invalidate (or upgrade it) from the related flow that has an allocation. let Some((ip, port)) = get_flow_masquerading_allocation(flow_info) else { return; }; + // Flows without VPC identity cannot be validated against the replacement config. let (Some(dst_vpcd), Some(src_vpcd)) = (flow_info.get_dst_vpcd(), flow_key.src_vpcd()) else { error!("Flow {flow_key} has no VPC discriminant, so it cannot be checked. This is a bug"); @@ -112,13 +116,15 @@ pub(crate) fn check_masquerading_flow( return; }; + // Check if there exists a peering with masquerading between the two VPCs of the flow debug!("Checking flow {}", flow_info.logfmt()); - let Some(nat_peering) = config.get_peering(src_vpcd, dst_vpcd) else { - debug!("Invalidating flow: there is no longer a peering for {src_vpcd} -- {dst_vpcd}"); + let Some(masq_peering) = config.find_masquerade_peering(src_vpcd, dst_vpcd) else { + debug!("Invalidating flow: there is no masquerading peering for {src_vpcd} -- {dst_vpcd}"); flow_info.invalidate_pair(); return; }; - debug!("Found peering between {src_vpcd} and {dst_vpcd}..."); + let pname = masq_peering.peering.name(); + debug!("Found peering between {src_vpcd} and {dst_vpcd}: {pname}"); // We've found a peering with masquerade between the VPCs that this flow is exchanged. // Check if such a peering has ANY expose with masquerading that includes the address currently @@ -127,19 +133,20 @@ pub(crate) fn check_masquerading_flow( let src_ip = flow_key.src_ip(); // source of flow let mut compatible_expose_found = false; let mut alloced_ip_valid = false; - for expose in nat_peering.peering.local().valexp() { + for expose in masq_peering.peering.local().valexp() { if let Some(nat) = expose.nat() && nat.is_masquerade() && nat.as_range.iter().any(|pfx| pfx.prefix().covers_addr(&ip)) { - debug!("Masquerading ip {ip} is allowed over peering {src_vpcd} -- {dst_vpcd}"); alloced_ip_valid = true; + debug!("Masquerade address {ip} is allowed over peering {pname}"); + if expose .ips() .iter() .any(|pfx| pfx.prefix().covers_addr(src_ip)) { - debug!("Flow source {src_ip} is still included in expose"); + debug!("Flow source {src_ip} is still allowed over peering {pname}"); compatible_expose_found = true; break; } @@ -147,12 +154,12 @@ pub(crate) fn check_masquerading_flow( } if !alloced_ip_valid { - debug!("Masquerade ip {ip} is no longer allowed over peering {src_vpcd} -- {dst_vpcd}"); + debug!("Masquerade ip {ip} is no longer allowed over peering {pname}"); flow_info.invalidate_pair(); return; } if !compatible_expose_found { - debug!("Flow is no longer valid for masquerading between {src_vpcd} -- {dst_vpcd}"); + debug!("Flow is no longer valid for masquerading over peering {pname}"); flow_info.invalidate_pair(); return; } @@ -166,26 +173,17 @@ pub(crate) fn check_masquerading_flow( } } -/// Move live NAT flows to a replacement allocator while blocking flow insertion. -pub(crate) fn reconcile_nat_flows<'a>( +/// Migrate active masquerading flows. Flows that get checked and retained, get their +/// ip/port reserved in the new allocator. +pub(crate) fn check_masquerading_flows<'a>( flow_table: &'a FlowTable, new_allocator: &NatAllocator, ) -> FlowTableReadGuard<'a> { let genid = new_allocator.genid(); debug!("CHECKING flows against new masquerade configuration with genid {genid}..."); let guard = flow_table.for_each_flow_filtered( - |_, f| f.is_active(), - |flow_key, flow_info| { - check_masquerading_flow(flow_key, flow_info, new_allocator); - // The check may invalidate a flow selected while it was still active. - if !flow_info.is_active() { - return; - } - if let Err(error) = update_port_forward_lease(flow_info, Some(new_allocator)) { - error!("Failed to reserve a live port-forward tuple: {error}"); - flow_info.invalidate_pair(); - } - }, + |_, f| f.is_active() && f.locked.read().nat_state.is_some(), + |flow_key, flow_info| check_masquerading_flow(flow_key, flow_info, new_allocator), ); debug!("CHECKING flows against new masquerade configuration COMPLETED"); guard diff --git a/nat/src/masquerade/mod.rs b/nat/src/masquerade/mod.rs index 968e942e5a..d7b2861dba 100644 --- a/nat/src/masquerade/mod.rs +++ b/nat/src/masquerade/mod.rs @@ -15,7 +15,6 @@ mod test; // re exports pub use allocator_writer::MasqueradeConfig; -pub(crate) use allocator_writer::NatAllocatorReader; pub use allocator_writer::NatAllocatorWriter; pub use nf::Masquerade; diff --git a/nat/src/portfw/flow_state.rs b/nat/src/portfw/flow_state.rs index efef56ee01..0b5c8c5677 100644 --- a/nat/src/portfw/flow_state.rs +++ b/nat/src/portfw/flow_state.rs @@ -6,7 +6,7 @@ #![allow(clippy::single_match_else)] use net::buffer::PacketBufferMut; -use net::flows::{ExtractMut, ExtractRef, FlowStatus}; +use net::flows::{ExtractRef, FlowStatus}; use net::ip::UnicastIpAddr; use net::packet::{Packet, VpcDiscriminant}; use net::{FlowKey, IpProtoKey}; @@ -19,72 +19,40 @@ use concurrency::sync::{Arc, Weak}; use flow_entry::flow_table::FlowInfo; use crate::common::{AtomicNatFlowStatus, NatAction, NatFlowStatus}; -use crate::masquerade::allocation::AllocatorError; -use crate::masquerade::apalloc::{Allocation, NatAllocator}; use crate::portfw::PortFwEntry; use crate::portfw::protocol::next_flow_status; #[allow(unused)] use tracing::{debug, error, warn}; -#[derive(Debug, Clone)] -pub(crate) struct PublicTuple { - ip: UnicastIpAddr, - port: NonZero, - peer_vpcd: VpcDiscriminant, - lease: Option>, -} - -impl PublicTuple { - pub(crate) fn new( - ip: UnicastIpAddr, - port: NonZero, - peer_vpcd: VpcDiscriminant, - lease: Option>, - ) -> Self { - Self { - ip, - port, - peer_vpcd, - lease, - } - } - - fn set_lease(&mut self, lease: Option>) { - self.lease = lease; - } -} - #[derive(Debug, Clone)] pub struct PortFwState { pub(crate) action: NatAction, pub(crate) status: AtomicNatFlowStatus, use_ip: UnicastIpAddr, use_port: NonZero, - public: PublicTuple, pub(crate) rule: Weak, } impl PortFwState { #[must_use] - pub(crate) fn new_snat( - public: PublicTuple, + pub fn new_snat( + use_ip: UnicastIpAddr, + use_port: NonZero, rule: Weak, status: AtomicNatFlowStatus, ) -> Self { Self { action: NatAction::SrcNat, status, - use_ip: public.ip, - use_port: public.port, - public, + use_ip, + use_port, rule, } } #[must_use] - pub(crate) fn new_dnat( + pub fn new_dnat( use_ip: UnicastIpAddr, use_port: NonZero, - public: PublicTuple, rule: Weak, status: AtomicNatFlowStatus, ) -> Self { @@ -93,7 +61,6 @@ impl PortFwState { status, use_ip, use_port, - public, rule, } } @@ -110,22 +77,6 @@ impl PortFwState { self.use_port } #[must_use] - pub(crate) fn public_ip(&self) -> UnicastIpAddr { - self.public.ip - } - #[must_use] - pub(crate) fn public_port(&self) -> NonZero { - self.public.port - } - pub(crate) fn set_lease(&mut self, lease: Option>) { - self.public.set_lease(lease); - } - #[cfg(test)] - #[must_use] - pub(crate) fn lease(&self) -> Option<&Arc> { - self.public.lease.as_ref() - } - #[must_use] pub fn rule(&self) -> &Weak { &self.rule } @@ -147,52 +98,6 @@ impl Display for PortFwState { } } -/// Update the public-tuple lease of a flow with port-forwarding state. -pub(crate) fn update_port_forward_lease( - flow_info: &FlowInfo, - allocator: Option<&NatAllocator>, -) -> Result<(), AllocatorError> { - let (public_ip, public_port, peer_vpcd) = { - let locked = flow_info.locked.read(); - let Some(state) = locked - .port_fw_state - .as_ref() - .and_then(|state| state.extract_ref::()) - else { - return Ok(()); - }; - ( - state.public_ip(), - state.public_port(), - state.public.peer_vpcd, - ) - }; - - let lease = allocator - .map(|allocator| { - allocator.reserve_port_forward( - flow_info.flowkey().proto(), - peer_vpcd, - public_ip.inner(), - public_port, - ) - }) - .transpose()? - .flatten(); - - let mut locked = flow_info.locked.write(); - let Some(state) = locked - .port_fw_state - .as_mut() - .and_then(|state| state.extract_mut::()) - else { - debug!("Port-forwarding state vanished while updating its lease"); - return Ok(()); - }; - state.set_lease(lease); - Ok(()) -} - // Build the flow keys for a port-forwarding flow pub(crate) fn build_portfw_flow_keys( packet: &mut Packet, // packet to be port-forwarded (in the forward path) @@ -231,14 +136,12 @@ pub(crate) fn setup_forward_flow( entry: &Arc, new_dst_ip: UnicastIpAddr, new_dst_port: NonZero, - public: PublicTuple, ) -> AtomicNatFlowStatus { // build port forwarding state for the forward flow let status = AtomicNatFlowStatus::new(); let port_fw_state = PortFwState::new_dnat( new_dst_ip, new_dst_port, - public, Arc::downgrade(entry), status.clone(), ); @@ -257,11 +160,12 @@ pub(crate) fn setup_reverse_flow( reverse_key: &FlowKey, reverse_flow: &Arc, entry: &Arc, - public: PublicTuple, + dst_ip: UnicastIpAddr, + dst_port: NonZero, status: AtomicNatFlowStatus, ) { // build port forwarding state for the REVERSE flow - let port_fw_state = PortFwState::new_snat(public, Arc::downgrade(entry), status); + let port_fw_state = PortFwState::new_snat(dst_ip, dst_port, Arc::downgrade(entry), status); // set the port forwarding state in the flow { diff --git a/nat/src/portfw/mod.rs b/nat/src/portfw/mod.rs index 3448499a54..20d367e319 100644 --- a/nat/src/portfw/mod.rs +++ b/nat/src/portfw/mod.rs @@ -13,7 +13,6 @@ mod test; // re-exports pub use flow_state::PortFwState; -pub(crate) use flow_state::update_port_forward_lease; pub use nf::PortForwarder; pub use portfwtable::PortFwTableError; pub use portfwtable::access::{PortFwTableReader, PortFwTableReaderFactory, PortFwTableWriter}; diff --git a/nat/src/portfw/nf.rs b/nat/src/portfw/nf.rs index 22a46b03ae..af5262236f 100644 --- a/nat/src/portfw/nf.rs +++ b/nat/src/portfw/nf.rs @@ -3,7 +3,6 @@ //! Port forwarding stage -use crate::masquerade::NatAllocatorReader; use crate::portfw::{PortFwEntry, PortFwKey, PortFwState, PortFwTable, PortFwTableReader}; use concurrency::sync::{Arc, Weak}; use flow_entry::flow_table::table::FlowTable; @@ -18,13 +17,11 @@ use std::num::NonZero; use std::time::Instant; use crate::common::NatAction; -use crate::portfw::flow_state::PublicTuple; use crate::portfw::flow_state::build_portfw_flow_keys; use crate::portfw::flow_state::get_packet_port_fw_state; use crate::portfw::flow_state::refresh_port_fw_entry; use crate::portfw::flow_state::setup_forward_flow; use crate::portfw::flow_state::setup_reverse_flow; -use crate::portfw::flow_state::update_port_forward_lease; use crate::portfw::packet::nat_packet; #[allow(unused)] @@ -35,24 +32,17 @@ pub struct PortForwarder { name: String, flow_table: Arc, fwtable: PortFwTableReader, - allocator: NatAllocatorReader, pipeline_data: Arc, } impl PortForwarder { /// Creates a new [`PortForwarder`] #[must_use] - pub fn new( - name: &str, - fwtable: PortFwTableReader, - flow_table: Arc, - allocator: NatAllocatorReader, - ) -> Self { + pub fn new(name: &str, fwtable: PortFwTableReader, flow_table: Arc) -> Self { Self { name: name.to_string(), flow_table, fwtable, - allocator, pipeline_data: Arc::from(PipelineData::default()), } } @@ -123,23 +113,6 @@ impl PortForwarder { return; }; - let lease = match self.allocator.get().map(|allocator| { - allocator.reserve_port_forward( - fw_key.proto(), - entry.key.src_vpcd(), - dst_ip.inner(), - dst_port, - ) - }) { - Some(Ok(lease)) => lease, - Some(Err(error)) => { - debug!("Unable to reserve {dst_ip}:{dst_port} for port forwarding: {error}"); - packet.done((&error).into()); - return; - } - None => None, - }; - // create a pair of related flow entries (outside the flow table). Timeout is set according to the rule matched let timeout = Instant::now() + entry.init_timeout(); let Ok((fw_flow, rev_flow)) = FlowInfo::related_pair( @@ -158,16 +131,8 @@ impl PortForwarder { fw_flow.set_genid_pair(self.pipeline_data.genid()); // set the flows in the FORWARD & REVERSE direction for subsequent packets - let public = PublicTuple::new(dst_ip, dst_port, entry.key.src_vpcd(), lease); - let status = setup_forward_flow( - &fw_key, - &fw_flow, - entry, - new_dst_ip, - new_dst_port, - public.clone(), - ); - setup_reverse_flow(&rev_key, &rev_flow, entry, public, status); + let status = setup_forward_flow(&fw_key, &fw_flow, entry, new_dst_ip, new_dst_port); + setup_reverse_flow(&rev_key, &rev_flow, entry, dst_ip, dst_port, status); // get the state we just created for the FORWARD direction let locked = fw_flow.locked.read(); @@ -203,14 +168,6 @@ impl PortForwarder { return; } - let allocator = self.allocator.get(); - for flow in [&fw_flow, &rev_flow] { - if let Err(error) = update_port_forward_lease(flow, allocator.as_deref()) { - flow.invalidate_pair(); - packet.done((&error).into()); - return; - } - } debug!("Inserted forward and reverse port-forwarding flow entries"); } diff --git a/nat/src/portfw/test.rs b/nat/src/portfw/test.rs index b2a62335eb..95c60451b1 100644 --- a/nat/src/portfw/test.rs +++ b/nat/src/portfw/test.rs @@ -3,17 +3,11 @@ #[cfg(test)] mod nf_test { - use crate::NatPort; use crate::common::NatFlowStatus; - use crate::masquerade::allocation::AllocatorError; - use crate::masquerade::apalloc::Allocation; - use crate::masquerade::{MasqueradeConfig, NatAllocatorWriter}; use crate::portfw::{PortForwarder, PortFwEntry, PortFwKey, PortFwState, PortFwTableWriter}; - use concurrency::sync::{Arc, Weak}; - use config::external::overlay::vpc::{Peering, ValidatedVpcTable, Vpc, VpcTable}; - use config::external::overlay::vpcpeering::{VpcExpose, VpcManifest}; - use flow_entry::flow_table::{FlowInfo, FlowLookup, FlowTable}; + use concurrency::sync::Arc; + use flow_entry::flow_table::{FlowLookup, FlowTable}; use lpm::prefix::Prefix; use net::buffer::TestBuffer; use net::flows::FlowStatus; @@ -23,8 +17,6 @@ mod nf_test { use net::packet::test_utils::{build_test_tcp_ipv4_packet, build_test_udp_ipv4_packet}; use net::packet::{DoneReason, Packet, VpcDiscriminant}; use pipeline::{DynPipeline, NetworkFunction}; - use std::net::IpAddr; - use std::num::NonZero; use std::str::FromStr; use std::time::Duration; use tracing_test::traced_test; @@ -137,51 +129,6 @@ mod nf_test { ruleset } - // Overlap the forwarded address with VPC-2's masquerade pool. - fn build_masquerade_vpc_table() -> ValidatedVpcTable { - let local = VpcManifest::with_exposes( - "VPC-2", - vec![ - VpcExpose::empty() - .make_masquerade(None) - .unwrap() - .ip("192.168.0.0/16".into()) - .as_range("70.71.72.0/24".into()) - .unwrap(), - ], - ); - let remote = - VpcManifest::with_exposes("VPC-1", vec![VpcExpose::empty().ip("10.0.0.0/24".into())]); - - let vpc1 = Vpc::new("VPC-1", "11111", 2000).unwrap(); - let mut vpc2 = Vpc::new("VPC-2", "22222", 3000).unwrap(); - vpc2.peerings.push(Peering { - name: "portfw_masquerade".into(), - local, - remote, - remote_id: "11111".try_into().unwrap(), - remote_vni: vpc1.vni, - gwgroup: "default".into(), - acl: None, - }); - - let mut vpctable = VpcTable::new(); - vpctable.add(vpc1).unwrap(); - vpctable.add(vpc2).unwrap(); - vpctable.validate().unwrap() - } - - const RELEASE_POLLS: usize = 64; - - fn lease_of(flow: &FlowInfo) -> Option> { - flow.locked - .read() - .port_fw_state - .as_ref() - .and_then(|state| state.extract_ref::()) - .and_then(|state| state.lease().cloned()) - } - // build a UDP packet to be port forwarded according to the port-forwarding table fn udp_packet_to_port_forward() -> Packet { let mut packet: Packet = @@ -249,20 +196,15 @@ mod nf_test { } } - fn setup_pipeline_with_allocator( + /// sets up a port-forwarding pipeline + fn setup_pipeline( ruleset: &[PortFwEntry], - allocator: &NatAllocatorWriter, ) -> (Arc, DynPipeline, PortFwTableWriter) { // build a pipeline with flow lookup + port forwarder let mut writer = PortFwTableWriter::new(); let flow_table = Arc::new(FlowTable::default()); let flow_lookup_nf = FlowLookup::new("flow-lookup", flow_table.clone()); - let nf = PortForwarder::new( - "port-forwarder", - writer.reader(), - flow_table.clone(), - allocator.get_reader(), - ); + let nf = PortForwarder::new("port-forwarder", writer.reader(), flow_table.clone()); let pipeline: DynPipeline = DynPipeline::new() .add_stage(flow_lookup_nf) .add_stage(TestFlowFilter) @@ -276,13 +218,6 @@ mod nf_test { (flow_table, pipeline, writer) } - fn setup_pipeline( - ruleset: &[PortFwEntry], - ) -> (Arc, DynPipeline, PortFwTableWriter) { - let allocator = NatAllocatorWriter::new(); - setup_pipeline_with_allocator(ruleset, &allocator) - } - #[cfg_attr(not(emulated), traced_test)] #[tokio::test] async fn test_nf_port_forwarding_base() { @@ -383,79 +318,6 @@ mod nf_test { ); } - #[cfg_attr(not(emulated), traced_test)] - #[tokio::test] - async fn port_forwarded_tuple_holds_a_masquerade_lease() { - let ruleset = build_test_port_forwarding_ruleset(); - let mut allocator = NatAllocatorWriter::new(); - let (flow_table, mut pipeline, _writer) = - setup_pipeline_with_allocator(&ruleset, &allocator); - allocator.update_nat_allocator( - MasqueradeConfig::new(&build_masquerade_vpc_table()), - 1, - &flow_table, - ); - - let output = process_packet(&mut pipeline, udp_packet_to_port_forward()); - assert!(!output.is_done()); - let reply = process_packet(&mut pipeline, build_reply(&output)); - - let one = reply - .meta() - .flow_info - .as_ref() - .expect("the reply matches the flow pair") - .clone(); - let other = one - .related - .as_ref() - .and_then(Weak::upgrade) - .expect("the forwarded flow has a related flow"); - - let lease_one = lease_of(&one).expect("the forwarded tuple overlaps a masquerade pool"); - let lease_other = lease_of(&other).expect("the forwarded tuple overlaps a masquerade pool"); - assert!( - Arc::ptr_eq(&lease_one, &lease_other), - "both directions must share one lease" - ); - let released = Arc::downgrade(&lease_one); - drop((lease_one, lease_other)); - - let nat = allocator - .get_reader() - .get() - .expect("allocator is installed"); - let port = NatPort::new_port(NonZero::new(3053).unwrap()); - let public: IpAddr = "70.71.72.73".parse().unwrap(); - let private: IpAddr = "192.168.1.2".parse().unwrap(); - - match nat.reserve_port(NextHeader::UDP, vpcd2(), vpcd1(), private, public, port) { - Err(AllocatorError::PortReservationFailed(blocked)) => assert_eq!(blocked, 3053), - other => panic!("a live lease must block the reservation, got {other:?}"), - } - - let keys = [*one.flowkey(), *other.flowkey()]; - drop((reply, output, one, other, pipeline)); - for key in &keys { - flow_table.remove(key).expect("the flow was in the table"); - } - - // Timer tasks retain flow state until cancellation is polled. - for _ in 0..RELEASE_POLLS { - if released.upgrade().is_none() { - break; - } - tokio::task::yield_now().await; - } - assert!( - released.upgrade().is_none(), - "retiring both flows must release the lease", - ); - - nat.reserve_port(NextHeader::UDP, vpcd2(), vpcd1(), private, public, port) - .expect("the tuple is free once the forwarded flows are gone"); - } - #[cfg_attr(not(emulated), traced_test)] #[tokio::test] async fn test_nf_port_forwarding_tcp_establishment() { diff --git a/nat/src/test.rs b/nat/src/test.rs index c8178f640b..6451c6399b 100644 --- a/nat/src/test.rs +++ b/nat/src/test.rs @@ -113,12 +113,7 @@ fn setup_masq_pipeline( portfw_writer .update_from_vpc_table(overlay.vpc_table()) .unwrap(); - let portfw = PortForwarder::new( - "port-forwarder", - portfw_writer.reader(), - flow_table.clone(), - allocator.get_reader(), - ); + let portfw = PortForwarder::new("port-forwarder", portfw_writer.reader(), flow_table.clone()); if let Some(table) = portfw_writer.enter() { println!("{}", table.as_ref()); } @@ -178,9 +173,11 @@ fn build_overlapping_masquerade_and_port_forward() -> ValidatedOverlay { Overlay::new(vpc_table, peerings).validate().unwrap() } +// The port-forwarding rule of that overlay claims 5.6.7.8:1024, the lowest port masquerade would +// otherwise hand out of the very same address. #[tokio::test] #[dpdk::with_eal] -async fn inactive_port_forward_does_not_reduce_masquerade_space() { +async fn a_port_forwarded_tuple_is_never_masqueraded() { let overlay = build_overlapping_masquerade_and_port_forward(); let (mut pipeline, flow_table, _flow_filter, _static_nat, _portfw, mut allocator) = setup_masq_pipeline(&overlay); @@ -194,12 +191,16 @@ async fn inactive_port_forward_does_not_reduce_masquerade_space() { let output: Vec<_> = pipeline.process(std::iter::once(packet)).collect(); let output = output.first().expect("masquerade should accept the flow"); assert_eq!(output.ip_source(), Some(addr("5.6.7.8"))); - assert_eq!(output.transport_src_port().unwrap().get(), 1024); + assert_ne!( + output.transport_src_port().unwrap().get(), + 1024, + "masquerade handed out the port a forwarding rule claims" + ); } #[tokio::test] #[dpdk::with_eal] -async fn active_port_forward_reserves_its_public_tuple() { +async fn a_claimed_tuple_cannot_be_reserved_for_masquerade() { let overlay = build_overlapping_masquerade_and_port_forward(); let (mut pipeline, flow_table, _flow_filter, _static_nat, _portfw, mut allocator) = setup_masq_pipeline(&overlay); @@ -209,6 +210,7 @@ async fn active_port_forward_reserves_its_public_tuple() { &flow_table, ); + // Several clients reach the forwarded service through the one claimed tuple. for (client, port) in [("1.2.3.4", 5000), ("1.2.3.5", 5001)] { let packet = build_packet(client, "5.6.7.8", port, 1024, vni(100)); let output: Vec<_> = pipeline.process(std::iter::once(packet)).collect(); @@ -219,7 +221,9 @@ async fn active_port_forward_reserves_its_public_tuple() { assert_eq!(output.transport_dst_port().unwrap().get(), 8000); } - let tuple_is_reserved = |allocator: &NatAllocatorWriter| { + // The claim holds whether or not a forwarded flow is using the tuple, and it survives allocator + // replacement because the replacement withholds it too. + let tuple_is_withheld = |allocator: &NatAllocatorWriter| { matches!( allocator.get_reader().get().unwrap().reserve_port( NextHeader::UDP, @@ -229,17 +233,17 @@ async fn active_port_forward_reserves_its_public_tuple() { addr("5.6.7.8"), NatPort::new_port_checked(1024).unwrap(), ), - Err(AllocatorError::PortReservationFailed(1024)) + Err(AllocatorError::Denied) ) }; - assert!(tuple_is_reserved(&allocator)); + assert!(tuple_is_withheld(&allocator)); allocator.update_nat_allocator( MasqueradeConfig::new(overlay.vpc_table()).set_randomize(true), 3, &flow_table, ); - assert!(tuple_is_reserved(&allocator)); + assert!(tuple_is_withheld(&allocator)); } #[tokio::test]