diff options
| author | Nataliia S. <81954790+natali-rs1985@users.noreply.github.com> | 2025-05-01 20:15:32 +0300 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2025-05-01 18:15:32 +0100 |
| commit | 528248af9628c3170b4eac399f0bb339072d8eae (patch) | |
| tree | 5b62b4734e42fd790f1c85ea5de350400dad1212 /src/conf_mode | |
| parent | f4e02fdff7a7fcb42a1c5fc313e317af63cd1dcd (diff) | |
| download | vyos-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-x | src/conf_mode/vpp.py | 2 | ||||
| -rw-r--r-- | src/conf_mode/vpp_nat.py | 12 | ||||
| -rw-r--r-- | src/conf_mode/vpp_nat_cgnat.py | 211 |
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) |
