Extend support for port forwarding with masquerading (stateful NAT)

config/src/converters/k8s/config/expose.rs

@@ -6,6 +6,7 @@ use std::convert::TryFrom;
 use k8s_intf::gateway_agent_crd::{
     GatewayAgentPeeringsPeeringExpose, GatewayAgentPeeringsPeeringExposeAs,
     GatewayAgentPeeringsPeeringExposeIps, GatewayAgentPeeringsPeeringExposeNat,
+    GatewayAgentPeeringsPeeringExposeNatPortForwardPortsProto,
 };
 use lpm::prefix::{PortRange, Prefix, PrefixString, PrefixWithOptionalPorts, PrefixWithPorts};
 
@@ -135,9 +136,9 @@ fn process_as_block(
 fn process_nat_block(
     vpc_expose: VpcExpose,
     nat: Option<&GatewayAgentPeeringsPeeringExposeNat>,
-) -> Result<VpcExpose, FromK8sConversionError> {
+) -> Result<Vec<VpcExpose>, FromK8sConversionError> {
     let Some(nat) = nat else {
-        return Ok(vpc_expose);
+        return Ok(vec![vpc_expose]);
     };
     match (&nat.masquerade, &nat.port_forward, &nat.r#static) {
         (Some(_), Some(_), _) | (Some(_), _, Some(_)) | (_, Some(_), Some(_)) => {
@@ -147,26 +148,137 @@ fn process_nat_block(
         }
         (Some(masquerade), None, None) => {
             let idle_timeout = masquerade.idle_timeout.map(std::time::Duration::from);
-            vpc_expose
-                .make_stateful_nat(idle_timeout)
+            Ok(vec![vpc_expose.make_stateful_nat(idle_timeout).map_err(
+                |e| FromK8sConversionError::NotAllowed(e.to_string()),
+            )?])
+        }
+        (None, Some(port_forward), None) => {
+            let idle_timeout = port_forward.idle_timeout.map(std::time::Duration::from);
+            let orig_port_ranges = if let Some(vec) = port_forward.ports.as_ref() {
+                Some(vec.iter().try_fold(Vec::new(), |mut acc, ports| {
+                    let Some(port_string) = ports.port.as_ref() else {
+                        return Err(FromK8sConversionError::NotAllowed(
+                            "Port forward must specify a port".to_string(),
+                        ));
+                    };
+                    let Some(as_string) = ports.r#as.as_ref() else {
+                        return Err(FromK8sConversionError::NotAllowed(
+                            "Port forward must specify a port to map to".to_string(),
+                        ));
+                    };
+                    acc.push((
+                        parse_port_ranges(port_string)?,
+                        parse_port_ranges(as_string)?,
+                        ports.proto.as_ref(),
+                    ));
+                    Ok(acc)
+                })?)
+            } else {
+                None
+            };
+            set_port_ranges(vpc_expose, orig_port_ranges)?
+                .into_iter()
+                .map(|vpc_expose| vpc_expose.make_port_forwarding(idle_timeout))
+                .collect::<Result<Vec<_>, _>>()
                 .map_err(|e| FromK8sConversionError::NotAllowed(e.to_string()))
         }
-        (None, Some(_port_forward), None) => {
-            // TODO: port forwarding support
-            // Don't forget to also update the TypeGenerator in k8s-intf/src/bolero/expose.rs
-            Err(FromK8sConversionError::NotAllowed(
-                "Port forwarding is not yet supported".to_string(),
-            ))
+        (None, None, Some(_)) => {
+            Ok(vec![vpc_expose.make_stateless_nat().map_err(|e| {
+                FromK8sConversionError::NotAllowed(e.to_string())
+            })?])
         }
-        (None, None, Some(_)) => vpc_expose
-            .make_stateless_nat()
-            .map_err(|e| FromK8sConversionError::NotAllowed(e.to_string())),
         (None, None, None) => Err(FromK8sConversionError::NotAllowed(
             "NAT config block must specify one NAT mode".to_string(),
         )),
     }
 }
-impl TryFrom<(&SubnetMap, &GatewayAgentPeeringsPeeringExpose)> for VpcExpose {
+
+#[allow(clippy::type_complexity)]
+fn set_port_ranges(
+    vpc_expose: VpcExpose,
+    orig_port_ranges: Option<
+        Vec<(
+            Vec<PortRange>,
+            Vec<PortRange>,
+            Option<&GatewayAgentPeeringsPeeringExposeNatPortForwardPortsProto>,
+        )>,
+    >,
+) -> Result<Vec<VpcExpose>, FromK8sConversionError> {
+    let Some(port_ranges_vec) = orig_port_ranges else {
+        return Ok(vec![vpc_expose]);
+    };
+    if port_ranges_vec.is_empty() {
+        return Err(FromK8sConversionError::NotAllowed(
+            "Port forward must specify at least one ports block".to_string(),
+        ));
+    }
+
+    let mut vpc_exposes = Vec::new();
+    for (orig_ranges, target_ranges, _proto) in port_ranges_vec {
+        let mut vpc_expose_clone = vpc_expose.clone();
+
+        let nat = vpc_expose_clone
+            .nat
+            .as_mut()
+            .unwrap_or_else(|| unreachable!());
+
+        let orig_prefixes = vpc_expose_clone.ips.clone();
+        let target_prefixes = nat.as_range.clone();
+
+        for orig_prefix in &orig_prefixes {
+            debug_assert!(orig_prefix.ports().is_none());
+            for range in &orig_ranges {
+                vpc_expose_clone.ips.insert(PrefixWithOptionalPorts::new(
+                    orig_prefix.prefix(),
+                    Some(*range),
+                ));
+            }
+            vpc_expose_clone.ips.remove(orig_prefix);
+        }
+
+        for target_prefix in &target_prefixes {
+            debug_assert!(target_prefix.ports().is_none());
+            for range in &target_ranges {
+                nat.as_range.insert(PrefixWithOptionalPorts::new(
+                    target_prefix.prefix(),
+                    Some(*range),
+                ));
+            }
+            nat.as_range.remove(target_prefix);
+        }
+        vpc_exposes.push(vpc_expose_clone);
+
+        // TODO: L4 protocol
+    }
+
+    Ok(vpc_exposes)
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct VpcExposes(Vec<VpcExpose>);
+
+impl VpcExposes {
+    #[cfg(test)]
+    fn get_single(self) -> Result<VpcExpose, FromK8sConversionError> {
+        if self.0.len() != 1 {
+            return Err(FromK8sConversionError::NotAllowed(
+                "Expected exactly one VPC expose".to_owned(),
+            ));
+        }
+        Ok(self.0.into_iter().next().unwrap())
+    }
+}
+
+impl IntoIterator for VpcExposes {
+    type Item = VpcExpose;
+    type IntoIter = std::vec::IntoIter<Self::Item>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.into_iter()
+    }
+}
+
+impl TryFrom<(&SubnetMap, &GatewayAgentPeeringsPeeringExpose)> for VpcExposes {
     type Error = FromK8sConversionError;
 
     fn try_from(
@@ -187,7 +299,7 @@ impl TryFrom<(&SubnetMap, &GatewayAgentPeeringsPeeringExpose)> for VpcExpose {
                     "A Default expose can't contain 'as' prefixes".to_string(),
                 ));
             }
-            return Ok(vpc_expose);
+            return Ok(VpcExposes(vec![vpc_expose]));
         }
 
         // Process PeeringIP rules
@@ -206,15 +318,15 @@ impl TryFrom<(&SubnetMap, &GatewayAgentPeeringsPeeringExpose)> for VpcExpose {
             ));
         }
 
-        vpc_expose = process_nat_block(vpc_expose, expose.nat.as_ref())?;
-
         if let Some(ases) = expose.r#as.as_ref() {
             for r#as in ases {
                 vpc_expose = process_as_block(vpc_expose, r#as)?;
             }
         }
 
-        Ok(vpc_expose)
+        let vpc_exposes = process_nat_block(vpc_expose, expose.nat.as_ref())?;
+
+        Ok(VpcExposes(vpc_exposes))
     }
 }
 
@@ -391,7 +503,7 @@ mod test {
         bolero::check!()
             .with_generator(expose_gen)
             .for_each(|k8s_expose| {
-                let expose = VpcExpose::try_from((&subnets, k8s_expose)).unwrap();
+                let expose = VpcExposes::try_from((&subnets, k8s_expose)).unwrap().get_single().unwrap();
                 let mut ips = expose
                     .ips
                     .iter()
@@ -507,6 +619,9 @@ mod test {
                             assert_eq!(config.idle_timeout, std::time::Duration::new(2 * 60, 0));
                         }
                     },
+                    (Some(VpcExposeNatConfig::PortForwarding(_)), Some(_)) => {
+                        todo!()
+                    }
                     (Some(VpcExposeNatConfig::Stateless(_)), Some(k8s_nat)) => {
                         assert!(k8s_nat.r#static.is_some(), "Stateless NAT configured but not by K8s: {expose:#?}\nk8s: {k8s_nat:#?}");
                     },

config/src/converters/k8s/config/peering.rs

@@ -1,11 +1,12 @@
 // SPDX-License-Identifier: Apache-2.0
 // Copyright Open Network Fabric Authors
 
+use super::expose::VpcExposes;
 use k8s_intf::gateway_agent_crd::{GatewayAgentPeerings, GatewayAgentPeeringsPeering};
 
 use crate::converters::k8s::FromK8sConversionError;
 use crate::converters::k8s::config::{SubnetMap, VpcSubnetMap};
-use crate::external::overlay::vpcpeering::{VpcExpose, VpcManifest, VpcPeering};
+use crate::external::overlay::vpcpeering::{VpcManifest, VpcPeering};
 
 impl TryFrom<(&SubnetMap, &str, &GatewayAgentPeeringsPeering)> for VpcManifest {
     type Error = FromK8sConversionError;
@@ -15,7 +16,7 @@ impl TryFrom<(&SubnetMap, &str, &GatewayAgentPeeringsPeering)> for VpcManifest {
     ) -> Result<Self, Self::Error> {
         let mut manifest = VpcManifest::new(vpc_name);
         for expose in peering.expose.as_ref().unwrap_or(&vec![]) {
-            manifest.add_expose(VpcExpose::try_from((subnets, expose))?);
+            manifest.add_exposes(VpcExposes::try_from((subnets, expose))?);
         }
         Ok(manifest)
     }

config/src/errors.rs

@@ -79,14 +79,8 @@ pub enum ConfigError {
     // NAT-specific
     #[error("Mismatched prefixes sizes for static NAT: {0:?} and {1:?}")]
     MismatchedPrefixSizes(PrefixWithPortsSize, PrefixWithPortsSize),
-    #[error(
-        "Stateful NAT is only supported on one side of a peering, but peering {0} has stateful NAT enabled on both sides"
-    )]
-    StatefulNatOnBothSides(String),
-    #[error(
-        "Stateful NAT is not compatible with stateless NAT at the other side of a peering, but peering {0} has stateful and stateless NAT enabled on different sides"
-    )]
-    StatefulPlusStatelessNat(String),
+    #[error("Peering {0} has manifests using incompatible NAT modes")]
+    IncompatibleNatModes(String),
 
     // Interface addresses
     #[error("Invalid interface address format: {0}")]

config/src/external/overlay/tests.rs

@@ -327,7 +327,7 @@ pub mod test {
         let mut overlay = Overlay::new(vpc_table, peering_table);
         assert_eq!(
             overlay.validate(),
-            Err(ConfigError::StatefulNatOnBothSides("Peering-1".to_owned()))
+            Err(ConfigError::IncompatibleNatModes("Peering-1".to_owned()))
         );
     }
 
@@ -373,9 +373,7 @@ pub mod test {
         let mut overlay = Overlay::new(vpc_table, peering_table);
         assert_eq!(
             overlay.validate(),
-            Err(ConfigError::StatefulPlusStatelessNat(
-                "Peering-1".to_owned()
-            ))
+            Err(ConfigError::IncompatibleNatModes("Peering-1".to_owned()))
         );
     }
 

config/src/external/overlay/vpc.rs

@@ -15,6 +15,7 @@ use tracing::{debug, error, warn};
 use crate::converters::k8s::config::peering;
 use crate::external::overlay::VpcManifest;
 use crate::external::overlay::VpcPeeringTable;
+use crate::external::overlay::vpcpeering::VpcExposeNatConfig;
 use crate::internal::interfaces::interface::{InterfaceConfig, InterfaceConfigTable};
 use crate::{ConfigError, ConfigResult};
 
@@ -51,29 +52,55 @@ impl Peering {
     fn validate_nat_combinations(&self) -> ConfigResult {
         // If stateful NAT is set up on one side of the peering, we don't support NAT (stateless or
         // stateful) on the other side.
-        let mut local_has_nat = false;
+        let mut local_has_stateless_nat = false;
         let mut local_has_stateful_nat = false;
+        let mut local_has_port_forwarding = false;
         for expose in &self.local.exposes {
-            if expose.has_stateful_nat() {
-                local_has_stateful_nat = true;
-                local_has_nat = true;
-                break;
-            } else if expose.has_stateless_nat() {
-                local_has_nat = true;
+            match expose.nat_config() {
+                Some(VpcExposeNatConfig::Stateful { .. }) => {
+                    local_has_stateful_nat = true;
+                }
+                Some(VpcExposeNatConfig::Stateless { .. }) => {
+                    local_has_stateless_nat = true;
+                }
+                Some(VpcExposeNatConfig::PortForwarding { .. }) => {
+                    local_has_port_forwarding = true;
+                }
+                None => {}
             }
         }
+        let local_has_nat =
+            local_has_stateless_nat || local_has_stateful_nat || local_has_port_forwarding;
 
         if !local_has_nat {
             return Ok(());
         }
 
-        for expose in &self.remote.exposes {
-            if expose.has_stateful_nat() {
-                return Err(ConfigError::StatefulNatOnBothSides(self.name.clone()));
+        let local_has_stateless_nat_only =
+            local_has_stateless_nat && !local_has_stateful_nat && !local_has_port_forwarding;
+
+        // Allowed:
+        //
+        // - no NAT ------------ *
+        // - stateless NAT ----- stateless NAT
+        //
+        // Disallowed (some of them may be supported in the future):
+        //
+        // - stateful NAT ------ stateless NAT
+        // - stateful NAT ------ stateful NAT
+        // - stateful NAT ------ port forwarding
+        // - port forwarding --- port forwarding
+        // - port forwarding --- stateless NAT
+
+        for remote_expose in &self.remote.exposes {
+            if !remote_expose.has_nat() {
+                continue;
             }
-            if expose.has_stateless_nat() && local_has_stateful_nat {
-                return Err(ConfigError::StatefulPlusStatelessNat(self.name.clone()));
+            if local_has_stateless_nat_only && remote_expose.has_stateless_nat() {
+                continue;
             }
+            // Other combinations are rejected
+            return Err(ConfigError::IncompatibleNatModes(self.name.clone()));
         }
         Ok(())
     }