decouple port forwarding and masquerading

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)

config/src/external/overlay/vpc.rs

@@ -310,7 +310,7 @@ impl Vpc {
             .map(Peering::validate)
             .collect::<Result<_, _>>()?;
 
-        let route_table = VpcRouteTable::build(&validated_peerings).validate()?;
+        let route_table = VpcRouteTable::build(&validated_peerings)?;
 
         let validated_vpc = ValidatedVpc {
             name: self.name.clone(),

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<ValidatedPeering>) -> Self {
+    ///
+    /// # Errors
+    ///
+    /// This function returns `ConfigError` if the `VpcRouteTable` does not
+    /// pass validation successfully
+    pub fn build(peerings: &Vec<ValidatedPeering>) -> Result<Self, ConfigError> {
         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<Self, ConfigError> {
+    fn validate(self) -> Result<Self, ConfigError> {
         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..] {

dataplane/src/packet_processor/mod.rs

@@ -122,7 +122,6 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
             "port-forwarder",
             portfw_factory.handle(),
             flow_table_clone.clone(),
-            natallocator_factory.handle(),
         );
         let pkt_stats_nf = PacketStatsNF::new(pkt_stats.clone());
 

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<FlowKey, Arc<FlowInfo>, RandomState>>,
 );

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<MasqueradePeering>,
@@ -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);
     }

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<I: NatIpWithBitmap> AllocatedIp<I> {
     fn new(
         ip: I,
         ip_allocator: IpAllocator<I>,
+        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<I: NatIpWithBitmap> AllocatedIp<I> {
         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<I> {
         &self.port_allocator
@@ -326,12 +338,19 @@ pub(crate) struct NatPool<I: NatIpWithBitmap> {
     bitmap_mapping: BTreeMap<u32, u128>,
     reverse_bitmap_mapping: BTreeMap<u128, u32>,
     in_use: VecDeque<Weak<AllocatedIp<I>>>,
+    /// 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<I: NatIpWithBitmap> NatPool<I> {
     /// 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<I: NatIpWithBitmap> NatPool<I> {
             bitmap_mapping,
             reverse_bitmap_mapping,
             in_use: VecDeque::new(),
+            reserved,
             exclude_wellknown_ports,
         }
     }
@@ -384,11 +404,16 @@ impl<I: NatIpWithBitmap> NatPool<I> {
         // 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<I: NatIpWithBitmap> NatPool<I> {
         // 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,
         ));

nat/src/masquerade/apalloc/concurrent_fuzz.rs

@@ -192,16 +192,18 @@ impl Scenario {
     fn specs(&self) -> Vec<PoolSpec> {
         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::<Ipv4Addr>(
             &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::<Ipv4Addr>(
             &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::<Ipv4Addr>(
             &specs,
             NextHeader::TCP,

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)?;