summaryrefslogtreecommitdiff
path: root/src/conf_mode
diff options
context:
space:
mode:
authorNataliia S. <81954790+natali-rs1985@users.noreply.github.com>2025-05-01 20:15:32 +0300
committerGitHub <noreply@github.com>2025-05-01 18:15:32 +0100
commit528248af9628c3170b4eac399f0bb339072d8eae (patch)
tree5b62b4734e42fd790f1c85ea5de350400dad1212 /src/conf_mode
parentf4e02fdff7a7fcb42a1c5fc313e317af63cd1dcd (diff)
downloadvyos-1x-528248af9628c3170b4eac399f0bb339072d8eae.tar.gz
vyos-1x-528248af9628c3170b4eac399f0bb339072d8eae.zip
T7390: VPP CGNAT implementation (#30)
CLI: ``` set vpp nat cgnat interface outside <interface> # multi set vpp nat cgnat interface inside <interface> # multi set vpp nat cgnat rule <rule> outside-prefix <prefix> set vpp nat cgnat rule <rule> inside-prefix <prefix> set vpp nat cgnat timeout udp <sec> # default 300 set vpp nat cgnat timeout tcp-established <sec> # default 7440 set vpp nat cgnat timeout tcp-transitory <sec> # default 240 set vpp nat cgnat timeout icmp <sec> # default 60 ``` OP mode: ``` show vpp nat cgnat interfaces show vpp nat cgnat mappings show vpp nat cgnat sessions clear vpp cgnat inside-address <address> port <port> external-address <address> port <port> ```
Diffstat (limited to 'src/conf_mode')
-rwxr-xr-xsrc/conf_mode/vpp.py2
-rw-r--r--src/conf_mode/vpp_nat.py12
-rw-r--r--src/conf_mode/vpp_nat_cgnat.py211
3 files changed, 223 insertions, 2 deletions
diff --git a/src/conf_mode/vpp.py b/src/conf_mode/vpp.py
index 7f9adc68b..3718f3897 100755
--- a/src/conf_mode/vpp.py
+++ b/src/conf_mode/vpp.py
@@ -138,6 +138,8 @@ def get_config(config=None):
# NAT dependency
if conf.exists(['vpp', 'nat44']):
set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
if not conf.exists(base):
return {
diff --git a/src/conf_mode/vpp_nat.py b/src/conf_mode/vpp_nat.py
index d38599388..a6bc70032 100644
--- a/src/conf_mode/vpp_nat.py
+++ b/src/conf_mode/vpp_nat.py
@@ -186,7 +186,11 @@ def verify(config):
raise ConfigError(
f'{interface} must be a VPP interface for "address-pool translation interface"'
)
- iface_address = get_interface_address(interface).get('addr_info', [])[0].get('local')
+ iface_address = (
+ get_interface_address(interface)
+ .get('addr_info', [])[0]
+ .get('local')
+ )
addresses_translation.append(iface_address)
if 'twice_nat' in address_pool:
@@ -211,7 +215,11 @@ def verify(config):
raise ConfigError(
f'{interface} must be a VPP interface for "address-pool twice-nat interface"'
)
- iface_address = get_interface_address(interface).get('addr_info', [])[0].get('local')
+ iface_address = (
+ get_interface_address(interface)
+ .get('addr_info', [])[0]
+ .get('local')
+ )
addresses_twice_nat.append(iface_address)
if 'static' in config:
diff --git a/src/conf_mode/vpp_nat_cgnat.py b/src/conf_mode/vpp_nat_cgnat.py
new file mode 100644
index 000000000..05260c476
--- /dev/null
+++ b/src/conf_mode/vpp_nat_cgnat.py
@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2025 VyOS Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with this program; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
+
+from vyos import ConfigError
+from vyos.config import Config, config_dict_merge
+from vyos.configdict import node_changed
+from vyos.configdiff import Diff
+from vyos.vpp.utils import cli_ifaces_list
+
+from vyos.vpp.nat.det44 import Det44
+
+
+def get_config(config=None) -> dict:
+ if config:
+ conf = config
+ else:
+ conf = Config()
+
+ base = ['vpp', 'nat', 'cgnat']
+
+ # Get config_dict with default values
+ config = conf.get_config_dict(
+ base,
+ key_mangling=('-', '_'),
+ get_first_key=True,
+ no_tag_node_value_mangle=True,
+ )
+
+ # Get effective config as we need full dictionary to delete
+ effective_config = conf.get_config_dict(
+ base,
+ key_mangling=('-', '_'),
+ effective=True,
+ get_first_key=True,
+ no_tag_node_value_mangle=True,
+ )
+
+ if not config:
+ config['remove'] = True
+
+ # Get default values which we need to conditionally update into the
+ # dictionary retrieved.
+ default_values = conf.get_config_defaults(**config.kwargs, recursive=True)
+ config = config_dict_merge(default_values, config)
+
+ config_changed = node_changed(
+ conf,
+ base,
+ key_mangling=('-', '_'),
+ recursive=True,
+ expand_nodes=Diff.DELETE | Diff.ADD,
+ )
+
+ changed_rules = node_changed(
+ conf,
+ base + ['rule'],
+ key_mangling=('-', '_'),
+ recursive=True,
+ expand_nodes=Diff.DELETE | Diff.ADD,
+ )
+
+ if not config_changed:
+ changed_rules = list(config.get('rule', {}).keys())
+
+ config.update(
+ {
+ 'changed_rules': changed_rules,
+ 'vpp_ifaces': cli_ifaces_list(conf),
+ }
+ )
+
+ if effective_config:
+ config.update({'effective': effective_config})
+
+ return config
+
+
+def verify(config):
+ if 'remove' in config:
+ return None
+
+ if 'interface' not in config:
+ raise ConfigError('Interfaces must be configured for CGNAT')
+ if 'rule' not in config:
+ raise ConfigError('Rules must be configured for CGNAT')
+
+ required_keys = {'inside', 'outside'}
+ missing_keys = required_keys - set(config['interface'].keys())
+ if missing_keys:
+ raise ConfigError(
+ f'Both inside and outside interfaces must be configured. '
+ f'Please add: {", ".join(missing_keys)}'
+ )
+
+ conflict_ifaces = set(config['interface']['inside']).intersection(
+ set(config['interface']['outside'])
+ )
+ if conflict_ifaces:
+ raise ConfigError(
+ f'Interface cannot be both inside and outside. '
+ f'Please choose a side for: {", ".join(conflict_ifaces)} '
+ )
+
+ for interface in config['interface']['inside']:
+ if interface not in config['vpp_ifaces']:
+ raise ConfigError(
+ f'{interface} must be a VPP interface for inside CGNAT interface'
+ )
+ for interface in config['interface']['outside']:
+ if interface not in config['vpp_ifaces']:
+ raise ConfigError(
+ f'{interface} must be a VPP interface for outside CGNAT interface'
+ )
+
+ required_keys = {'outside_prefix', 'inside_prefix'}
+ for rule in config['rule']:
+ missing_keys = required_keys - set(config['rule'][rule].keys())
+ if missing_keys:
+ raise ConfigError(
+ f'Both inside-prefix and outside-prefix must be configured in rule {rule}. '
+ f'Please add: {", ".join(missing_keys).replace("_", "-")}'
+ )
+
+
+def generate(config):
+ pass
+
+
+def apply(config):
+ cgnat = Det44()
+
+ if 'remove' in config:
+ cgnat.disable_det44_plugin()
+ return None
+
+ if 'effective' in config:
+ remove_config = config.get('effective')
+ # Delete inside interfaces
+ for interface in remove_config['interface']['inside']:
+ if interface not in config.get('interface', {}).get('inside', []):
+ cgnat.delete_det44_interface_inside(interface)
+ # Delete outside interfaces
+ for interface in remove_config['interface']['outside']:
+ if interface not in config.get('interface', {}).get('outside', []):
+ cgnat.delete_det44_interface_outside(interface)
+ # Delete CGNAT rules
+ for rule in config['changed_rules']:
+ if rule in remove_config.get('rule', {}):
+ rule_config = remove_config['rule'][rule]
+ in_addr, in_plen = rule_config['inside_prefix'].split('/')
+ out_addr, out_plen = rule_config['outside_prefix'].split('/')
+ cgnat.delete_det44_mapping(
+ in_addr=in_addr,
+ in_plen=int(in_plen),
+ out_addr=out_addr,
+ out_plen=int(out_plen),
+ )
+
+ # Add DET44
+ cgnat.enable_det44_plugin()
+ # Add inside interfaces
+ for interface in config['interface']['inside']:
+ cgnat.add_det44_interface_inside(interface)
+ # Add outside interfaces
+ for interface in config['interface']['outside']:
+ cgnat.add_det44_interface_outside(interface)
+ # Add CGNAT rules
+ for rule in config['changed_rules']:
+ if rule in config.get('rule', {}):
+ rule_config = config['rule'][rule]
+ in_addr, in_plen = rule_config['inside_prefix'].split('/')
+ out_addr, out_plen = rule_config['outside_prefix'].split('/')
+ cgnat.add_det44_mapping(
+ in_addr=in_addr,
+ in_plen=int(in_plen),
+ out_addr=out_addr,
+ out_plen=int(out_plen),
+ )
+ # Set CGNAT timeouts
+ cgnat.set_det44_timeouts(
+ icmp=int(config['timeout']['icmp']),
+ udp=int(config['timeout']['udp']),
+ tcp_established=int(config['timeout']['tcp_established']),
+ tcp_transitory=int(config['timeout']['tcp_transitory']),
+ )
+
+
+if __name__ == '__main__':
+ try:
+ c = get_config()
+ verify(c)
+ generate(c)
+ apply(c)
+ except ConfigError as e:
+ print(e)
+ exit(1)