From 9bfe2749a1fa41a4c77a3364851ac1fbefbf3ef3 Mon Sep 17 00:00:00 2001 From: Nataliia Solomko Date: Fri, 13 Feb 2026 12:29:00 +0200 Subject: vpp: T8254: Move 'nat44' and 'settings nat44' sections to 'nat nat44' --- src/conf_mode/interfaces_ethernet.py | 10 +- src/conf_mode/vpp.py | 27 +- src/conf_mode/vpp_interfaces_bonding.py | 4 +- src/conf_mode/vpp_interfaces_gre.py | 4 +- src/conf_mode/vpp_interfaces_ipip.py | 4 +- src/conf_mode/vpp_interfaces_loopback.py | 4 +- src/conf_mode/vpp_interfaces_vxlan.py | 4 +- src/conf_mode/vpp_kernel-interfaces.py | 4 +- src/conf_mode/vpp_nat.py | 515 ------------------------------- src/conf_mode/vpp_nat_nat44.py | 513 ++++++++++++++++++++++++++++++ src/migration-scripts/vpp/7-to-8 | 34 ++ src/op_mode/show_vpp_nat44.py | 251 --------------- src/op_mode/vpp_nat_nat44.py | 251 +++++++++++++++ 13 files changed, 817 insertions(+), 808 deletions(-) delete mode 100644 src/conf_mode/vpp_nat.py create mode 100644 src/conf_mode/vpp_nat_nat44.py delete mode 100644 src/op_mode/show_vpp_nat44.py create mode 100644 src/op_mode/vpp_nat_nat44.py (limited to 'src') diff --git a/src/conf_mode/interfaces_ethernet.py b/src/conf_mode/interfaces_ethernet.py index 97902c6e4..774a042fe 100755 --- a/src/conf_mode/interfaces_ethernet.py +++ b/src/conf_mode/interfaces_ethernet.py @@ -357,12 +357,12 @@ def verify_vpp_remove_vif(ethernet: dict): # Known paths that already use VLAN interfaces r'(nat\.cgnat\.interface\.inside)|' r'(nat\.cgnat\.interface\.outside)|' - r'(nat44\.interface\.inside)|' - r'(nat44\.interface\.outside)|' + r'(nat\.nat44\.interface\.inside)|' + r'(nat\.nat44\.interface\.outside)|' # Potential paths for VLAN interfaces - r'(nat44\.address_pool\.translation\.interface)|' - r'(nat44\.address_pool\.twice_nat\.interface)|' - r'(nat44\.exclude\.rule\.(\d)+\.external_interface)|' + r'(nat\.nat44\.address_pool\.translation\.interface)|' + r'(nat\.nat44\.address_pool\.twice_nat\.interface)|' + r'(nat\.nat44\.exclude\.rule\.(\d)+\.external_interface)|' r'(interfaces\.bonding\.bond(\d)+\.member\.interface)|' r'(interfaces\.bridge\.br(\d)+\.member\.interface)|' r'(interfaces\.xconnect\.xcon(\d)+\.member\.interface)|' diff --git a/src/conf_mode/vpp.py b/src/conf_mode/vpp.py index ae75594d6..b26388cdc 100755 --- a/src/conf_mode/vpp.py +++ b/src/conf_mode/vpp.py @@ -54,7 +54,6 @@ from vyos.vpp.config_verify import ( verify_vpp_cpu_main_core, verify_vpp_settings_cpu_skip_cores, verify_vpp_settings_cpu_workers, - verify_vpp_nat44_workers, verify_vpp_memory, verify_vpp_statseg_size, verify_vpp_interfaces_dpdk_num_queues, @@ -442,8 +441,8 @@ def get_config(config=None): set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) @@ -539,11 +538,6 @@ def verify(config): workers = _get_workers_count(config['settings'].get('cpu', {})) - if 'workers' in config['settings']['nat44']: - verify_vpp_nat44_workers( - workers=workers, nat44_workers=config['settings']['nat44']['workers'] - ) - verify_vpp_main_heap_size(config['settings']) verify_vpp_statseg_size(config['settings']) @@ -882,23 +876,6 @@ def apply(config): # Syncronize routes via LCP vpp_control.lcp_resync() - # NAT44 settings - nat44_settings = config['settings'].get('nat44', {}) - - vpp_control.set_nat44_session_limit( - int(nat44_settings.get('session_limit')) - ) - - if nat44_settings.get('workers'): - bitmask = 0 - for worker_range in nat44_settings['workers']: - worker_numbers = worker_range.split('-') - for wid in range( - int(worker_numbers[0]), int(worker_numbers[-1]) + 1 - ): - bitmask |= 1 << wid - vpp_control.set_nat_workers(bitmask) - except (VPPIOError, VPPValueError, VppNotRunningError) as e: # if cannot connect to VPP or an error occurred then # we need to stop vpp service and initialize interfaces diff --git a/src/conf_mode/vpp_interfaces_bonding.py b/src/conf_mode/vpp_interfaces_bonding.py index d25d64eb0..23d0720bf 100644 --- a/src/conf_mode/vpp_interfaces_bonding.py +++ b/src/conf_mode/vpp_interfaces_bonding.py @@ -144,8 +144,8 @@ def get_config(config=None) -> dict: set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) diff --git a/src/conf_mode/vpp_interfaces_gre.py b/src/conf_mode/vpp_interfaces_gre.py index 0b60d06b5..53a423dd9 100644 --- a/src/conf_mode/vpp_interfaces_gre.py +++ b/src/conf_mode/vpp_interfaces_gre.py @@ -115,8 +115,8 @@ def get_config(config=None) -> dict: set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) diff --git a/src/conf_mode/vpp_interfaces_ipip.py b/src/conf_mode/vpp_interfaces_ipip.py index e4db1e93c..eae51f5a2 100644 --- a/src/conf_mode/vpp_interfaces_ipip.py +++ b/src/conf_mode/vpp_interfaces_ipip.py @@ -114,8 +114,8 @@ def get_config(config=None) -> dict: set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) diff --git a/src/conf_mode/vpp_interfaces_loopback.py b/src/conf_mode/vpp_interfaces_loopback.py index 8874ecf2d..d08557959 100644 --- a/src/conf_mode/vpp_interfaces_loopback.py +++ b/src/conf_mode/vpp_interfaces_loopback.py @@ -104,8 +104,8 @@ def get_config(config=None) -> dict: set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) diff --git a/src/conf_mode/vpp_interfaces_vxlan.py b/src/conf_mode/vpp_interfaces_vxlan.py index 355139ffa..4a0b157e5 100644 --- a/src/conf_mode/vpp_interfaces_vxlan.py +++ b/src/conf_mode/vpp_interfaces_vxlan.py @@ -121,8 +121,8 @@ def get_config(config=None) -> dict: set_dependents('vpp_kernel_interface', conf, iface) # NAT dependency - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) diff --git a/src/conf_mode/vpp_kernel-interfaces.py b/src/conf_mode/vpp_kernel-interfaces.py index fcbc8ecc0..2e0e630e5 100644 --- a/src/conf_mode/vpp_kernel-interfaces.py +++ b/src/conf_mode/vpp_kernel-interfaces.py @@ -82,8 +82,8 @@ def get_config(config=None) -> dict: if conf.exists(['vpp', 'nat', 'cgnat']): set_dependents('vpp_nat_cgnat', conf) - if conf.exists(['vpp', 'nat44']): - set_dependents('vpp_nat', conf) + if conf.exists(['vpp', 'nat', 'nat44']): + set_dependents('vpp_nat_nat44', conf) config['ifname'] = ifname diff --git a/src/conf_mode/vpp_nat.py b/src/conf_mode/vpp_nat.py deleted file mode 100644 index cfe527ab8..000000000 --- a/src/conf_mode/vpp_nat.py +++ /dev/null @@ -1,515 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 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. - -import ipaddress - -from vyos import ConfigError - -from vyos.configdiff import Diff -from vyos.configdict import node_changed -from vyos.config import Config -from vyos.utils.network import get_interface_address - -from vyos.vpp.utils import cli_ifaces_list -from vyos.vpp.utils import vpp_iface_name_transform -from vyos.vpp.nat.nat44 import Nat44 -from vyos.vpp.control_vpp import VPPControl - - -protocol_map = { - 'all': 0, - 'icmp': 1, - 'tcp': 6, - 'udp': 17, -} - - -def get_config(config=None) -> dict: - if config: - conf = config - else: - conf = Config() - - base = ['vpp', 'nat44'] - - # Get config_dict with default values - config = conf.get_config_dict( - base, - key_mangling=('-', '_'), - get_first_key=True, - no_tag_node_value_mangle=True, - with_defaults=True, - with_recursive_defaults=True, - ) - - if not conf.exists(['vpp']): - config['remove_vpp'] = True - return config - - # Get effective config as we need full dictionary for deletion - 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 - return config - - config_changed = node_changed( - conf, - base, - key_mangling=('-', '_'), - recursive=True, - expand_nodes=Diff.DELETE | Diff.ADD, - ) - - changed_static_rules = node_changed( - conf, - base + ['static', 'rule'], - key_mangling=('-', '_'), - recursive=True, - expand_nodes=Diff.DELETE | Diff.ADD, - ) - - changed_exclude_rules = node_changed( - conf, - base + ['exclude', 'rule'], - key_mangling=('-', '_'), - recursive=True, - expand_nodes=Diff.DELETE | Diff.ADD, - ) - - if not config_changed: - changed_static_rules = list(config.get('static', {}).get('rule', {}).keys()) - changed_exclude_rules = list(config.get('exclude', {}).get('rule', {}).keys()) - - config.update( - { - 'changed_static_rules': changed_static_rules, - 'changed_exclude_rules': changed_exclude_rules, - 'vpp_ifaces': cli_ifaces_list(conf), - } - ) - - settings = conf.get_config_dict( - ['vpp', 'settings', 'nat44'], - key_mangling=('-', '_'), - with_recursive_defaults=True, - ) - config.update(settings.get('nat44')) - - if effective_config: - config.update({'effective': effective_config}) - - return config - - -def convert_range_to_list_ips(address_range) -> list: - """Converts IP range to a list of IPs . - - Example: - % ip = IPOperations('192.0.0.1-192.0.2.5') - % ip.convert_prefix_to_list_ips() - ['192.0.2.1', '192.0.2.2', '192.0.2.3', '192.0.2.4', '192.0.2.5'] - """ - if '-' in address_range: - start_ip, end_ip = address_range.split('-') - start_ip = ipaddress.ip_address(start_ip) - end_ip = ipaddress.ip_address(end_ip) - return [ - str(ipaddress.ip_address(ip)) - for ip in range(int(start_ip), int(end_ip) + 1) - ] - else: - return [address_range] - - -def verify(config): - if 'remove' in config or 'remove_vpp' in config: - return None - - if 'interface' not in config: - raise ConfigError('Interfaces must be configured for NAT44') - - 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. Please add: {", ".join(missing_keys)}' - ) - - vpp = VPPControl() - for direction in ['inside', 'outside']: - for interface in config['interface'][direction]: - vpp_iface_name = vpp_iface_name_transform(interface) - if vpp.get_sw_if_index(vpp_iface_name) is None: - raise ConfigError( - f'{interface} must be a VPP interface for {direction} NAT interface' - ) - - if not config.get('address_pool', {}).get('translation') and not config.get( - 'static', {} - ).get('rule'): - raise ConfigError('"address-pool translation" or "static rule" is required') - - addresses_translation = [] - addresses_twice_nat = [] - if 'address_pool' in config: - address_pool = config.get('address_pool') - if 'translation' in address_pool: - if not address_pool['translation'].get('address') and not address_pool[ - 'translation' - ].get('interface'): - raise ConfigError( - '"address-pool translation" requires address or interface' - ) - - for address_range in address_pool['translation'].get('address', []): - addresses = convert_range_to_list_ips(address_range) - for address in addresses: - if address in addresses_translation: - raise ConfigError( - f'Address {address} is already in use in "address-pool translation address"' - ) - addresses_translation.append(address) - - for interface in address_pool['translation'].get('interface', []): - if interface not in config['vpp_ifaces']: - raise ConfigError( - f'{interface} must be a VPP interface for "address-pool translation interface"' - ) - address_info = get_interface_address(interface).get('addr_info') - if not address_info: - raise ConfigError( - f'{interface} should have an address to be used for "address-pool translation interface"' - ) - iface_address = address_info[0].get('local') - addresses_translation.append(iface_address) - - if 'twice_nat' in address_pool: - if not address_pool['twice_nat'].get('address') and not address_pool[ - 'twice_nat' - ].get('interface'): - raise ConfigError( - '"address-pool twice-nat" requires address or interface' - ) - - for address_range in address_pool['twice_nat'].get('address', []): - addresses = convert_range_to_list_ips(address_range) - for address in addresses: - if address in addresses_twice_nat: - raise ConfigError( - f'Address {address} is already in use in "address-pool twice-nat address"' - ) - addresses_twice_nat.append(address) - - for interface in address_pool['twice_nat'].get('interface', []): - if interface not in config['vpp_ifaces']: - raise ConfigError( - f'{interface} must be a VPP interface for "address-pool twice-nat interface"' - ) - address_info = get_interface_address(interface).get('addr_info') - if not address_info: - raise ConfigError( - f'{interface} should have an address to be used for "address-pool twice-nat interface"' - ) - iface_address = address_info[0].get('local') - addresses_twice_nat.append(iface_address) - - if 'static' in config: - addresses_with_ports = set() - addresses_without_ports = set() - local_addresses = set() - - for rule, rule_config in config['static'].get('rule', {}).items(): - error_msg = f'Configuration error in static rule {rule}:' - - if not rule_config.get('local', {}).get('address'): - raise ConfigError(f'{error_msg} local settings require address') - - if not rule_config.get('external', {}).get('address'): - raise ConfigError(f'{error_msg} external settings require address') - - has_local_port = 'port' in rule_config.get('local', {}) - has_external_port = 'port' in rule_config.get('external', {}) - - if not has_external_port == has_local_port: - raise ConfigError( - f'{error_msg} source and destination ports must either ' - 'both be specified, or neither must be specified' - ) - - # Either both protocol and ports are set, or both no protocol and no ports - if (rule_config['protocol'] != 'all') != has_local_port: - raise ConfigError( - f'{error_msg} protocol and ports must either both be specified or both omitted' - ) - - ext_address = rule_config['external']['address'] - port = rule_config['external'].get('port') - local_address = rule_config['local']['address'] - - if port: - pair = (ext_address, port) - if ( - pair in addresses_with_ports - or ext_address in addresses_without_ports - ): - raise ConfigError( - f'{error_msg} external address/port is already in use!' - ) - addresses_with_ports.add(pair) - - else: - if ext_address in addresses_without_ports or any( - addr == ext_address for addr, _ in addresses_with_ports - ): - raise ConfigError( - f'{error_msg} external address is already in use!' - ) - addresses_without_ports.add(ext_address) - - if local_address in local_addresses: - raise ConfigError( - f'{error_msg} local address {local_address} is already in use' - ) - local_addresses.add(local_address) - - options = rule_config.get('options', {}) - - if 'self_twice_nat' in options and ext_address not in addresses_translation: - raise ConfigError( - f'{error_msg} external address {ext_address} must be part of ' - '"address-pool translation" when using self-twice-nat' - ) - - if all(key in options for key in ('twice_nat', 'self_twice_nat')): - raise ConfigError( - f'{error_msg} cannot set both options "twice-nat" and "self-twice-nat"' - ) - if any(key in options for key in ('twice_nat', 'self_twice_nat')): - if not has_local_port or rule_config['protocol'] == 'all': - raise ConfigError( - f'{error_msg} twice-nat/self-twice-nat options require port and protocol to be set' - ) - if not config.get('address_pool', {}).get('twice_nat'): - raise ConfigError( - f'{error_msg} twice-nat/self-twice-nat options require "address-pool twice-nat" to be set' - ) - if 'twice_nat_address' in options: - if not any(key in options for key in ('twice_nat', 'self_twice_nat')): - raise ConfigError( - f'{error_msg} twice-nat/self-twice-nat option required when twice-nat-address is set' - ) - tn_address = options['twice_nat_address'] - if tn_address not in addresses_twice_nat: - raise ConfigError( - f'{error_msg} twice-nat-address {tn_address} is not in "address-pool twice-nat"' - ) - - if 'exclude' in config: - for rule, rule_config in config['exclude'].get('rule', {}).items(): - keys = {'local_address', 'external_interface'} - if not any(key in rule_config for key in keys): - raise ConfigError( - f'Local-address or external-interface must be specified for exclude rule {rule}' - ) - if all(key in rule_config for key in keys): - raise ConfigError( - f'Cannot set both address and interface for exclude rule {rule}' - ) - if ( - 'external_interface' in rule_config - and rule_config.get('external_interface') not in config['vpp_ifaces'] - ): - raise ConfigError( - f'{rule_config["external_interface"]} must be a VPP interface for exclude rule {rule}' - ) - - # Either both protocol and local-port are set, or both no protocol and no port - if (rule_config['protocol'] != 'all') != ('local_port' in rule_config): - raise ConfigError( - f'Protocol and local-port must either both be specified or both omitted for exclude rule {rule}' - ) - - -def generate(config): - pass - - -def apply(config): - if 'remove_vpp' in config: - return None - - n = Nat44() - - if 'remove' in config: - n.disable_nat44_ed() - 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', []): - vpp_iface_name = vpp_iface_name_transform(interface) - n.delete_nat44_interface_inside(vpp_iface_name) - # Delete outside interfaces - for interface in remove_config['interface']['outside']: - if interface not in config.get('interface', {}).get('outside', []): - vpp_iface_name = vpp_iface_name_transform(interface) - n.delete_nat44_interface_outside(vpp_iface_name) - # Delete address pool - address_pool = config.get('address_pool', {}) - for address in ( - remove_config.get('address_pool', {}) - .get('translation', {}) - .get('address', []) - ): - if address not in address_pool.get('translation', {}).get('address', []): - n.delete_nat44_address_range(address, twice_nat=False) - for interface in ( - remove_config.get('address_pool', {}) - .get('translation', {}) - .get('interface', []) - ): - if interface not in address_pool.get('translation', {}).get( - 'interface', [] - ): - n.delete_nat44_interface_address(interface, twice_nat=False) - for address in ( - remove_config.get('address_pool', {}) - .get('twice_nat', {}) - .get('address', []) - ): - if address not in address_pool.get('twice_nat', {}).get('address', []): - n.delete_nat44_address_range(address, twice_nat=True) - for interface in ( - remove_config.get('address_pool', {}) - .get('twice_nat', {}) - .get('interface', []) - ): - if interface not in address_pool.get('twice_nat', {}).get('interface', []): - n.delete_nat44_interface_address(interface, twice_nat=True) - # Delete NAT static mapping rules - for rule in config['changed_static_rules']: - if rule in remove_config.get('static', {}).get('rule', {}): - rule_config = remove_config['static']['rule'][rule] - n.delete_nat44_static_mapping( - local_ip=rule_config.get('local').get('address'), - external_ip=rule_config.get('external', {}).get('address', ''), - local_port=int(rule_config.get('local', {}).get('port', 0)), - external_port=int(rule_config.get('external', {}).get('port', 0)), - protocol=protocol_map[rule_config.get('protocol', 'all')], - twice_nat='twice_nat' in rule_config.get('options', {}), - self_twice_nat='self_twice_nat' in rule_config.get('options', {}), - out2in='out_to_in_only' in rule_config.get('options', {}), - pool_ip=rule_config.get('options', {}).get('twice_nat_address'), - ) - # Delete NAT exclude rules - for rule in config['changed_exclude_rules']: - if rule in remove_config.get('exclude', {}).get('rule', {}): - rule_config = remove_config['exclude']['rule'][rule] - n.delete_nat44_identity_mapping( - ip_address=rule_config.get('local_address'), - protocol=protocol_map[rule_config.get('protocol', 'all')], - port=int(rule_config.get('local_port', 0)), - interface=rule_config.get('external_interface'), - ) - - # Add NAT44 - n.enable_nat44_ed() - - # Dynamic rules always require `address-pool translation` in CLI - we can use this for an easy validation - # Forwarding must be disabled when dynamic rules are present - # Without dynamic rules, forwarding remains enabled - enable_forwarding = not bool(config.get('address_pool', {}).get('translation')) - n.enable_disable_nat44_forwarding(enable_forwarding) - - # Add inside interfaces - for interface in config['interface']['inside']: - vpp_iface_name = vpp_iface_name_transform(interface) - n.add_nat44_interface_inside(vpp_iface_name) - # Add outside interfaces - for interface in config['interface']['outside']: - vpp_iface_name = vpp_iface_name_transform(interface) - n.add_nat44_interface_outside(vpp_iface_name) - # Add translation pool - for address in ( - config.get('address_pool', {}).get('translation', {}).get('address', []) - ): - n.add_nat44_address_range(address, twice_nat=False) - for interface in ( - config.get('address_pool', {}).get('translation', {}).get('interface', []) - ): - n.add_nat44_interface_address(interface, twice_nat=False) - for address in ( - config.get('address_pool', {}).get('twice_nat', {}).get('address', []) - ): - n.add_nat44_address_range(address, twice_nat=True) - for interface in ( - config.get('address_pool', {}).get('twice_nat', {}).get('interface', []) - ): - n.add_nat44_interface_address(interface, twice_nat=True) - # Add NAT static mapping rules - for rule in config['changed_static_rules']: - if rule in config.get('static', {}).get('rule', {}): - rule_config = config['static']['rule'][rule] - n.add_nat44_static_mapping( - local_ip=rule_config.get('local').get('address'), - external_ip=rule_config.get('external', {}).get('address', ''), - local_port=int(rule_config.get('local', {}).get('port', 0)), - external_port=int(rule_config.get('external', {}).get('port', 0)), - protocol=protocol_map[rule_config.get('protocol', 'all')], - twice_nat='twice_nat' in rule_config.get('options', {}), - self_twice_nat='self_twice_nat' in rule_config.get('options', {}), - out2in='out_to_in_only' in rule_config.get('options', {}), - pool_ip=rule_config.get('options', {}).get('twice_nat_address'), - ) - # Add NAT exclude rules - for rule in config['changed_exclude_rules']: - if rule in config.get('exclude', {}).get('rule', {}): - rule_config = config['exclude']['rule'][rule] - n.add_nat44_identity_mapping( - ip_address=rule_config.get('local_address'), - protocol=protocol_map[rule_config.get('protocol', 'all')], - port=int(rule_config.get('local_port', 0)), - interface=rule_config.get('external_interface'), - ) - if 'timeout' in config: - n.set_nat_timeouts( - icmp=int(config.get('timeout').get('icmp')), - udp=int(config.get('timeout').get('udp')), - tcp_established=int(config.get('timeout').get('tcp_established')), - tcp_transitory=int(config.get('timeout').get('tcp_transitory')), - ) - - -if __name__ == '__main__': - try: - c = get_config() - verify(c) - generate(c) - apply(c) - except ConfigError as e: - print(e) - exit(1) diff --git a/src/conf_mode/vpp_nat_nat44.py b/src/conf_mode/vpp_nat_nat44.py new file mode 100644 index 000000000..30111bf40 --- /dev/null +++ b/src/conf_mode/vpp_nat_nat44.py @@ -0,0 +1,513 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 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. + +import ipaddress + +from vyos import ConfigError + +from vyos.configdiff import Diff +from vyos.configdict import node_changed +from vyos.config import Config, config_dict_merge +from vyos.utils.network import get_interface_address + +from vyos.vpp.utils import cli_ifaces_list +from vyos.vpp.utils import vpp_iface_name_transform +from vyos.vpp.nat.nat44 import Nat44 +from vyos.vpp.control_vpp import VPPControl + + +protocol_map = { + 'all': 0, + 'icmp': 1, + 'tcp': 6, + 'udp': 17, +} + + +def get_config(config=None) -> dict: + if config: + conf = config + else: + conf = Config() + + base = ['vpp', 'nat', 'nat44'] + + # Get config_dict + config = conf.get_config_dict( + base, + key_mangling=('-', '_'), + get_first_key=True, + no_tag_node_value_mangle=True, + ) + + if not conf.exists(['vpp']): + config['remove_vpp'] = True + return config + + # Get effective config as we need full dictionary for deletion + 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 + return config + + # 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_static_rules = node_changed( + conf, + base + ['static', 'rule'], + key_mangling=('-', '_'), + recursive=True, + expand_nodes=Diff.DELETE | Diff.ADD, + ) + + changed_exclude_rules = node_changed( + conf, + base + ['exclude', 'rule'], + key_mangling=('-', '_'), + recursive=True, + expand_nodes=Diff.DELETE | Diff.ADD, + ) + + if not config_changed: + changed_static_rules = list(config.get('static', {}).get('rule', {}).keys()) + changed_exclude_rules = list(config.get('exclude', {}).get('rule', {}).keys()) + + config.update( + { + 'changed_static_rules': changed_static_rules, + 'changed_exclude_rules': changed_exclude_rules, + 'vpp_ifaces': cli_ifaces_list(conf), + } + ) + + if effective_config: + config.update({'effective': effective_config}) + + return config + + +def convert_range_to_list_ips(address_range) -> list: + """Converts IP range to a list of IPs . + + Example: + % ip = IPOperations('192.0.0.1-192.0.2.5') + % ip.convert_prefix_to_list_ips() + ['192.0.2.1', '192.0.2.2', '192.0.2.3', '192.0.2.4', '192.0.2.5'] + """ + if '-' in address_range: + start_ip, end_ip = address_range.split('-') + start_ip = ipaddress.ip_address(start_ip) + end_ip = ipaddress.ip_address(end_ip) + return [ + str(ipaddress.ip_address(ip)) + for ip in range(int(start_ip), int(end_ip) + 1) + ] + else: + return [address_range] + + +def verify(config): + if 'remove' in config or 'remove_vpp' in config: + return None + + if 'interface' not in config: + raise ConfigError('Interfaces must be configured for NAT44') + + 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. Please add: {", ".join(missing_keys)}' + ) + + vpp = VPPControl() + for direction in ['inside', 'outside']: + for interface in config['interface'][direction]: + vpp_iface_name = vpp_iface_name_transform(interface) + if vpp.get_sw_if_index(vpp_iface_name) is None: + raise ConfigError( + f'{interface} must be a VPP interface for {direction} NAT interface' + ) + + if not config.get('address_pool', {}).get('translation') and not config.get( + 'static', {} + ).get('rule'): + raise ConfigError('"address-pool translation" or "static rule" is required') + + addresses_translation = [] + addresses_twice_nat = [] + if 'address_pool' in config: + address_pool = config.get('address_pool') + if 'translation' in address_pool: + if not address_pool['translation'].get('address') and not address_pool[ + 'translation' + ].get('interface'): + raise ConfigError( + '"address-pool translation" requires address or interface' + ) + + for address_range in address_pool['translation'].get('address', []): + addresses = convert_range_to_list_ips(address_range) + for address in addresses: + if address in addresses_translation: + raise ConfigError( + f'Address {address} is already in use in "address-pool translation address"' + ) + addresses_translation.append(address) + + for interface in address_pool['translation'].get('interface', []): + if interface not in config['vpp_ifaces']: + raise ConfigError( + f'{interface} must be a VPP interface for "address-pool translation interface"' + ) + address_info = get_interface_address(interface).get('addr_info') + if not address_info: + raise ConfigError( + f'{interface} should have an address to be used for "address-pool translation interface"' + ) + iface_address = address_info[0].get('local') + addresses_translation.append(iface_address) + + if 'twice_nat' in address_pool: + if not address_pool['twice_nat'].get('address') and not address_pool[ + 'twice_nat' + ].get('interface'): + raise ConfigError( + '"address-pool twice-nat" requires address or interface' + ) + + for address_range in address_pool['twice_nat'].get('address', []): + addresses = convert_range_to_list_ips(address_range) + for address in addresses: + if address in addresses_twice_nat: + raise ConfigError( + f'Address {address} is already in use in "address-pool twice-nat address"' + ) + addresses_twice_nat.append(address) + + for interface in address_pool['twice_nat'].get('interface', []): + if interface not in config['vpp_ifaces']: + raise ConfigError( + f'{interface} must be a VPP interface for "address-pool twice-nat interface"' + ) + address_info = get_interface_address(interface).get('addr_info') + if not address_info: + raise ConfigError( + f'{interface} should have an address to be used for "address-pool twice-nat interface"' + ) + iface_address = address_info[0].get('local') + addresses_twice_nat.append(iface_address) + + if 'static' in config: + addresses_with_ports = set() + addresses_without_ports = set() + local_addresses = set() + + for rule, rule_config in config['static'].get('rule', {}).items(): + error_msg = f'Configuration error in static rule {rule}:' + + if not rule_config.get('local', {}).get('address'): + raise ConfigError(f'{error_msg} local settings require address') + + if not rule_config.get('external', {}).get('address'): + raise ConfigError(f'{error_msg} external settings require address') + + has_local_port = 'port' in rule_config.get('local', {}) + has_external_port = 'port' in rule_config.get('external', {}) + + if not has_external_port == has_local_port: + raise ConfigError( + f'{error_msg} source and destination ports must either ' + 'both be specified, or neither must be specified' + ) + + # Either both protocol and ports are set, or both no protocol and no ports + if (rule_config['protocol'] != 'all') != has_local_port: + raise ConfigError( + f'{error_msg} protocol and ports must either both be specified or both omitted' + ) + + ext_address = rule_config['external']['address'] + port = rule_config['external'].get('port') + local_address = rule_config['local']['address'] + + if port: + pair = (ext_address, port) + if ( + pair in addresses_with_ports + or ext_address in addresses_without_ports + ): + raise ConfigError( + f'{error_msg} external address/port is already in use!' + ) + addresses_with_ports.add(pair) + + else: + if ext_address in addresses_without_ports or any( + addr == ext_address for addr, _ in addresses_with_ports + ): + raise ConfigError( + f'{error_msg} external address is already in use!' + ) + addresses_without_ports.add(ext_address) + + if local_address in local_addresses: + raise ConfigError( + f'{error_msg} local address {local_address} is already in use' + ) + local_addresses.add(local_address) + + options = rule_config.get('options', {}) + + if 'self_twice_nat' in options and ext_address not in addresses_translation: + raise ConfigError( + f'{error_msg} external address {ext_address} must be part of ' + '"address-pool translation" when using self-twice-nat' + ) + + if all(key in options for key in ('twice_nat', 'self_twice_nat')): + raise ConfigError( + f'{error_msg} cannot set both options "twice-nat" and "self-twice-nat"' + ) + if any(key in options for key in ('twice_nat', 'self_twice_nat')): + if not has_local_port or rule_config['protocol'] == 'all': + raise ConfigError( + f'{error_msg} twice-nat/self-twice-nat options require port and protocol to be set' + ) + if not config.get('address_pool', {}).get('twice_nat'): + raise ConfigError( + f'{error_msg} twice-nat/self-twice-nat options require "address-pool twice-nat" to be set' + ) + if 'twice_nat_address' in options: + if not any(key in options for key in ('twice_nat', 'self_twice_nat')): + raise ConfigError( + f'{error_msg} twice-nat/self-twice-nat option required when twice-nat-address is set' + ) + tn_address = options['twice_nat_address'] + if tn_address not in addresses_twice_nat: + raise ConfigError( + f'{error_msg} twice-nat-address {tn_address} is not in "address-pool twice-nat"' + ) + + if 'exclude' in config: + for rule, rule_config in config['exclude'].get('rule', {}).items(): + keys = {'local_address', 'external_interface'} + if not any(key in rule_config for key in keys): + raise ConfigError( + f'Local-address or external-interface must be specified for exclude rule {rule}' + ) + if all(key in rule_config for key in keys): + raise ConfigError( + f'Cannot set both address and interface for exclude rule {rule}' + ) + if ( + 'external_interface' in rule_config + and rule_config.get('external_interface') not in config['vpp_ifaces'] + ): + raise ConfigError( + f'{rule_config["external_interface"]} must be a VPP interface for exclude rule {rule}' + ) + + # Either both protocol and local-port are set, or both no protocol and no port + if (rule_config['protocol'] != 'all') != ('local_port' in rule_config): + raise ConfigError( + f'Protocol and local-port must either both be specified or both omitted for exclude rule {rule}' + ) + + +def generate(config): + pass + + +def apply(config): + if 'remove_vpp' in config: + return None + + n = Nat44() + + if 'remove' in config: + n.disable_nat44_ed() + 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', []): + vpp_iface_name = vpp_iface_name_transform(interface) + n.delete_nat44_interface_inside(vpp_iface_name) + # Delete outside interfaces + for interface in remove_config['interface']['outside']: + if interface not in config.get('interface', {}).get('outside', []): + vpp_iface_name = vpp_iface_name_transform(interface) + n.delete_nat44_interface_outside(vpp_iface_name) + # Delete address pool + address_pool = config.get('address_pool', {}) + for address in ( + remove_config.get('address_pool', {}) + .get('translation', {}) + .get('address', []) + ): + if address not in address_pool.get('translation', {}).get('address', []): + n.delete_nat44_address_range(address, twice_nat=False) + for interface in ( + remove_config.get('address_pool', {}) + .get('translation', {}) + .get('interface', []) + ): + if interface not in address_pool.get('translation', {}).get( + 'interface', [] + ): + n.delete_nat44_interface_address(interface, twice_nat=False) + for address in ( + remove_config.get('address_pool', {}) + .get('twice_nat', {}) + .get('address', []) + ): + if address not in address_pool.get('twice_nat', {}).get('address', []): + n.delete_nat44_address_range(address, twice_nat=True) + for interface in ( + remove_config.get('address_pool', {}) + .get('twice_nat', {}) + .get('interface', []) + ): + if interface not in address_pool.get('twice_nat', {}).get('interface', []): + n.delete_nat44_interface_address(interface, twice_nat=True) + # Delete NAT static mapping rules + for rule in config['changed_static_rules']: + if rule in remove_config.get('static', {}).get('rule', {}): + rule_config = remove_config['static']['rule'][rule] + n.delete_nat44_static_mapping( + local_ip=rule_config.get('local').get('address'), + external_ip=rule_config.get('external', {}).get('address', ''), + local_port=int(rule_config.get('local', {}).get('port', 0)), + external_port=int(rule_config.get('external', {}).get('port', 0)), + protocol=protocol_map[rule_config.get('protocol', 'all')], + twice_nat='twice_nat' in rule_config.get('options', {}), + self_twice_nat='self_twice_nat' in rule_config.get('options', {}), + out2in='out_to_in_only' in rule_config.get('options', {}), + pool_ip=rule_config.get('options', {}).get('twice_nat_address'), + ) + # Delete NAT exclude rules + for rule in config['changed_exclude_rules']: + if rule in remove_config.get('exclude', {}).get('rule', {}): + rule_config = remove_config['exclude']['rule'][rule] + n.delete_nat44_identity_mapping( + ip_address=rule_config.get('local_address'), + protocol=protocol_map[rule_config.get('protocol', 'all')], + port=int(rule_config.get('local_port', 0)), + interface=rule_config.get('external_interface'), + ) + + # Add NAT44 + n.enable_nat44_ed() + + # Dynamic rules always require `address-pool translation` in CLI - we can use this for an easy validation + # Forwarding must be disabled when dynamic rules are present + # Without dynamic rules, forwarding remains enabled + enable_forwarding = not bool(config.get('address_pool', {}).get('translation')) + n.enable_disable_nat44_forwarding(enable_forwarding) + + # Add inside interfaces + for interface in config['interface']['inside']: + vpp_iface_name = vpp_iface_name_transform(interface) + n.add_nat44_interface_inside(vpp_iface_name) + # Add outside interfaces + for interface in config['interface']['outside']: + vpp_iface_name = vpp_iface_name_transform(interface) + n.add_nat44_interface_outside(vpp_iface_name) + # Add translation pool + for address in ( + config.get('address_pool', {}).get('translation', {}).get('address', []) + ): + n.add_nat44_address_range(address, twice_nat=False) + for interface in ( + config.get('address_pool', {}).get('translation', {}).get('interface', []) + ): + n.add_nat44_interface_address(interface, twice_nat=False) + for address in ( + config.get('address_pool', {}).get('twice_nat', {}).get('address', []) + ): + n.add_nat44_address_range(address, twice_nat=True) + for interface in ( + config.get('address_pool', {}).get('twice_nat', {}).get('interface', []) + ): + n.add_nat44_interface_address(interface, twice_nat=True) + # Add NAT static mapping rules + for rule in config['changed_static_rules']: + if rule in config.get('static', {}).get('rule', {}): + rule_config = config['static']['rule'][rule] + n.add_nat44_static_mapping( + local_ip=rule_config.get('local').get('address'), + external_ip=rule_config.get('external', {}).get('address', ''), + local_port=int(rule_config.get('local', {}).get('port', 0)), + external_port=int(rule_config.get('external', {}).get('port', 0)), + protocol=protocol_map[rule_config.get('protocol', 'all')], + twice_nat='twice_nat' in rule_config.get('options', {}), + self_twice_nat='self_twice_nat' in rule_config.get('options', {}), + out2in='out_to_in_only' in rule_config.get('options', {}), + pool_ip=rule_config.get('options', {}).get('twice_nat_address'), + ) + # Add NAT exclude rules + for rule in config['changed_exclude_rules']: + if rule in config.get('exclude', {}).get('rule', {}): + rule_config = config['exclude']['rule'][rule] + n.add_nat44_identity_mapping( + ip_address=rule_config.get('local_address'), + protocol=protocol_map[rule_config.get('protocol', 'all')], + port=int(rule_config.get('local_port', 0)), + interface=rule_config.get('external_interface'), + ) + if 'timeout' in config: + n.set_nat_timeouts( + icmp=int(config.get('timeout').get('icmp')), + udp=int(config.get('timeout').get('udp')), + tcp_established=int(config.get('timeout').get('tcp_established')), + tcp_transitory=int(config.get('timeout').get('tcp_transitory')), + ) + if 'session_limit' in config: + n.set_nat44_session_limit(int(config['session_limit'])) + + +if __name__ == '__main__': + try: + c = get_config() + verify(c) + generate(c) + apply(c) + except ConfigError as e: + print(e) + exit(1) diff --git a/src/migration-scripts/vpp/7-to-8 b/src/migration-scripts/vpp/7-to-8 index dfe2f5013..e6cff9668 100644 --- a/src/migration-scripts/vpp/7-to-8 +++ b/src/migration-scripts/vpp/7-to-8 @@ -16,6 +16,9 @@ # Rename `vpp settings logging default-log-level` to # `vpp settings logging default-level` (T8255) # +# Move 'vpp nat44' and 'vpp settings nat44' to 'vpp nat nat44'. +# Drop settings for nat workers (T8254) + from vyos.configtree import ConfigTree @@ -26,5 +29,36 @@ def _migrate_vpp_log(config: ConfigTree) -> None: config.rename(base + ['default-log-level'], 'default-level') +def _migrate_vpp_nat44(config: ConfigTree) -> None: + base_path = ['vpp', 'nat44'] + new_base_path = ['vpp', 'nat', 'nat44'] + settings_path = ['vpp', 'settings', 'nat44'] + + if not config.exists(base_path): + # Nothing to do + return + + if not config.exists(['vpp', 'nat']): + config.set(['vpp', 'nat']) + + # copy "vpp nat44" to "vpp nat nat44" + config.copy(base_path, new_base_path) + config.delete(base_path) + + # move 'vpp settings nat44' to 'vpp nat nat44' + if config.exists(settings_path): + if config.exists(settings_path + ['session-limit']): + session_limit = config.return_value(settings_path + ['session-limit']) + config.set(new_base_path + ['session-limit'], value=session_limit) + if config.exists(settings_path + ['timeout']): + config.copy(settings_path + ['timeout'], new_base_path + ['timeout']) + config.delete(settings_path) + + def migrate(config: ConfigTree) -> None: + if not config.exists(['vpp']): + # Nothing to do + return + _migrate_vpp_log(config) + _migrate_vpp_nat44(config) diff --git a/src/op_mode/show_vpp_nat44.py b/src/op_mode/show_vpp_nat44.py deleted file mode 100644 index 500280210..000000000 --- a/src/op_mode/show_vpp_nat44.py +++ /dev/null @@ -1,251 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 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. - -import json -import sys -from tabulate import tabulate - -import vyos.opmode -from vyos.configquery import ConfigTreeQuery - -from vyos.vpp import VPPControl - - -protocol_map = { - 0: 'all', - 1: 'icmp', - 6: 'tcp', - 17: 'udp', -} - -# NAT flags -flags_map = { - 'twice-nat': 0x01, - 'self-twice-nat': 0x02, - 'out2in-only': 0x04, - 'out': 0x10, - 'in': 0x20, -} - - -def _verify(func): - """Decorator checks if config for VPP NAT44 exists""" - from functools import wraps - - @wraps(func) - def _wrapper(*args, **kwargs): - config = ConfigTreeQuery() - base = 'vpp nat44' - if not config.exists(base): - raise vyos.opmode.UnconfiguredSubsystem(f'{base} is not configured') - - return func(*args, **kwargs) - - return _wrapper - - -def decode_bitmask(bitmask: int) -> list: - """Decode a bitmask into a list of flag names""" - return [name for name, value in flags_map.items() if bitmask & value] - - -def _get_raw_output(data_dump): - data = [json.loads(json.dumps(d._asdict(), default=str)) for d in data_dump] - return data - - -def _get_raw_output_sessions(vpp_api): - users: list[dict] = vpp_api.nat44_user_dump() - sessions_list: list[dict] = [] - for user in users: - ip_address = str(user._asdict().get('ip_address')) - user_sessions_dump = vpp_api.nat44_user_session_v3_dump(ip_address=ip_address) - user_sessions = [ - json.loads(json.dumps(session._asdict(), default=str)) - for session in user_sessions_dump - ] - sessions_list.extend(user_sessions) - return sorted(sessions_list, key=lambda x: x["inside_ip_address"]) - - -def _get_formatted_output_sessions(sessions_list): - print('NAT44 ED sessions:') - print(f'--------------- {len(sessions_list)} sessions ---------------') - for session in sessions_list: - in_ip_addr = session.get('inside_ip_address') - in_port = session.get('inside_port') - out_ip_addr = session.get('outside_ip_address') - out_port = session.get('outside_port') - protocol = protocol_map[session.get('protocol')].upper() - last_heard = session.get('last_heard') - time_since_last_heard = session.get('time_since_last_heard') - total_bytes = session.get('total_bytes') - total_pkts = session.get('total_pkts') - ext_host_address = session.get('ext_host_address') - ext_host_port = session.get('ext_host_port') - is_timed_out = session.get('is_timed_out') - - print(f' i2o {in_ip_addr} proto {protocol} port {in_port}') - print(f' o2i {out_ip_addr} proto {protocol} port {out_port}') - print(f' external host {ext_host_address}:{ext_host_port}') - print( - f' i2o flow: match: saddr {in_ip_addr} sport {in_port} daddr {ext_host_address} dport {ext_host_port} proto {protocol} rewrite: saddr {out_ip_addr}' - + ( - f' sport {out_port}' - if protocol != 'ICMP' - else f' daddr {ext_host_address} icmp-id {ext_host_port}' - ) - ) - print( - f' o2i flow: match: saddr {ext_host_address} sport {ext_host_port} daddr {out_ip_addr} dport {out_port} proto {protocol} rewrite: ' - + ( - f'daddr {in_ip_addr} dport {in_port}' - if protocol != 'ICMP' - else f' saddr {ext_host_address} daddr {in_ip_addr} icmp-id {ext_host_port}' - ) - ) - print(f' last heard {last_heard}') - print(f' time since last heard {time_since_last_heard}') - print(f' total packets {total_pkts}, total bytes {total_bytes}') - if is_timed_out: - print(' session timed out') - print('\n') - - -def _get_formatted_output_addresses(addresses): - twice_nat_address = [] - translation_address = [] - for address_info in addresses: - address = address_info.get('ip_address') - if address_info.get('flags') & flags_map['twice-nat']: - twice_nat_address.append(address) - else: - translation_address.append(address) - - print('NAT44 pool addresses:') - for addr in translation_address: - print(f' {addr}') - print('NAT44 twice-nat pool addresses:') - for addr in twice_nat_address: - print(f' {addr}') - - -def _get_formatted_output_interfaces(vpp, interfaces): - print('NAT44 interfaces:') - for interface in interfaces: - name = vpp.get_interface_name(interface['sw_if_index']) - iface_type = decode_bitmask(interface['flags']) - print(f' {name} {" ".join(iface_type)}') - - -def _get_formatted_output_rules(rules_list): - data_entries = [] - for rule in rules_list: - external_address = rule.get('external_ip_address') - external_port = rule.get('external_port') or '' - local_address = rule.get('local_ip_address') - local_port = rule.get('local_port') or '' - protocol = protocol_map[rule.get('protocol', 0)] - options = ' '.join(decode_bitmask(rule.get('flags'))) - - values = [ - external_address, - external_port, - local_address, - local_port, - protocol, - options, - ] - data_entries.append(values) - headers = [ - 'External address', - 'External port', - 'Local address', - 'Local port', - 'Protocol', - 'Options', - ] - out = sorted(data_entries, key=lambda x: x[2]) - return tabulate(out, headers=headers, tablefmt='simple') - - -@_verify -def show_sessions(raw: bool): - vpp = VPPControl() - sessions_list: list[dict] = _get_raw_output_sessions(vpp.api) - - if raw: - return sessions_list - - else: - return _get_formatted_output_sessions(sessions_list) - - -@_verify -def show_summary(raw: bool): - vpp = VPPControl() - return vpp.cli_cmd('show nat44 summary').reply - - -@_verify -def show_static(raw: bool): - vpp = VPPControl() - nat_static_dump = vpp.api.nat44_static_mapping_dump() - rules_list: list[dict] = _get_raw_output(nat_static_dump) - - if raw: - return rules_list - - else: - return _get_formatted_output_rules(rules_list) - - -@_verify -def show_addresses(raw: bool): - vpp = VPPControl() - addresses_dump = vpp.api.nat44_address_dump() - addresses: list[dict] = _get_raw_output(addresses_dump) - - if raw: - return addresses - - else: - return _get_formatted_output_addresses(addresses) - - -@_verify -def show_interfaces(raw: bool): - vpp = VPPControl() - interfaces_dump = vpp.api.nat44_interface_dump() - interfaces: list[dict] = _get_raw_output(interfaces_dump) - - if raw: - return interfaces - - else: - return _get_formatted_output_interfaces(vpp, interfaces) - - -if __name__ == '__main__': - try: - res = vyos.opmode.run(sys.modules[__name__]) - if res: - print(res) - except (ValueError, vyos.opmode.Error) as e: - print(e) - sys.exit(1) diff --git a/src/op_mode/vpp_nat_nat44.py b/src/op_mode/vpp_nat_nat44.py new file mode 100644 index 000000000..97fca8dbc --- /dev/null +++ b/src/op_mode/vpp_nat_nat44.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 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. + +import json +import sys +from tabulate import tabulate + +import vyos.opmode +from vyos.configquery import ConfigTreeQuery + +from vyos.vpp import VPPControl + + +protocol_map = { + 0: 'all', + 1: 'icmp', + 6: 'tcp', + 17: 'udp', +} + +# NAT flags +flags_map = { + 'twice-nat': 0x01, + 'self-twice-nat': 0x02, + 'out2in-only': 0x04, + 'out': 0x10, + 'in': 0x20, +} + + +def _verify(func): + """Decorator checks if config for VPP NAT44 exists""" + from functools import wraps + + @wraps(func) + def _wrapper(*args, **kwargs): + config = ConfigTreeQuery() + base = 'vpp nat nat44' + if not config.exists(base): + raise vyos.opmode.UnconfiguredSubsystem(f'{base} is not configured') + + return func(*args, **kwargs) + + return _wrapper + + +def decode_bitmask(bitmask: int) -> list: + """Decode a bitmask into a list of flag names""" + return [name for name, value in flags_map.items() if bitmask & value] + + +def _get_raw_output(data_dump): + data = [json.loads(json.dumps(d._asdict(), default=str)) for d in data_dump] + return data + + +def _get_raw_output_sessions(vpp_api): + users: list[dict] = vpp_api.nat44_user_dump() + sessions_list: list[dict] = [] + for user in users: + ip_address = str(user._asdict().get('ip_address')) + user_sessions_dump = vpp_api.nat44_user_session_v3_dump(ip_address=ip_address) + user_sessions = [ + json.loads(json.dumps(session._asdict(), default=str)) + for session in user_sessions_dump + ] + sessions_list.extend(user_sessions) + return sorted(sessions_list, key=lambda x: x["inside_ip_address"]) + + +def _get_formatted_output_sessions(sessions_list): + print('NAT44 ED sessions:') + print(f'--------------- {len(sessions_list)} sessions ---------------') + for session in sessions_list: + in_ip_addr = session.get('inside_ip_address') + in_port = session.get('inside_port') + out_ip_addr = session.get('outside_ip_address') + out_port = session.get('outside_port') + protocol = protocol_map[session.get('protocol')].upper() + last_heard = session.get('last_heard') + time_since_last_heard = session.get('time_since_last_heard') + total_bytes = session.get('total_bytes') + total_pkts = session.get('total_pkts') + ext_host_address = session.get('ext_host_address') + ext_host_port = session.get('ext_host_port') + is_timed_out = session.get('is_timed_out') + + print(f' i2o {in_ip_addr} proto {protocol} port {in_port}') + print(f' o2i {out_ip_addr} proto {protocol} port {out_port}') + print(f' external host {ext_host_address}:{ext_host_port}') + print( + f' i2o flow: match: saddr {in_ip_addr} sport {in_port} daddr {ext_host_address} dport {ext_host_port} proto {protocol} rewrite: saddr {out_ip_addr}' + + ( + f' sport {out_port}' + if protocol != 'ICMP' + else f' daddr {ext_host_address} icmp-id {ext_host_port}' + ) + ) + print( + f' o2i flow: match: saddr {ext_host_address} sport {ext_host_port} daddr {out_ip_addr} dport {out_port} proto {protocol} rewrite: ' + + ( + f'daddr {in_ip_addr} dport {in_port}' + if protocol != 'ICMP' + else f' saddr {ext_host_address} daddr {in_ip_addr} icmp-id {ext_host_port}' + ) + ) + print(f' last heard {last_heard}') + print(f' time since last heard {time_since_last_heard}') + print(f' total packets {total_pkts}, total bytes {total_bytes}') + if is_timed_out: + print(' session timed out') + print('\n') + + +def _get_formatted_output_addresses(addresses): + twice_nat_address = [] + translation_address = [] + for address_info in addresses: + address = address_info.get('ip_address') + if address_info.get('flags') & flags_map['twice-nat']: + twice_nat_address.append(address) + else: + translation_address.append(address) + + print('NAT44 pool addresses:') + for addr in translation_address: + print(f' {addr}') + print('NAT44 twice-nat pool addresses:') + for addr in twice_nat_address: + print(f' {addr}') + + +def _get_formatted_output_interfaces(vpp, interfaces): + print('NAT44 interfaces:') + for interface in interfaces: + name = vpp.get_interface_name(interface['sw_if_index']) + iface_type = decode_bitmask(interface['flags']) + print(f' {name} {" ".join(iface_type)}') + + +def _get_formatted_output_rules(rules_list): + data_entries = [] + for rule in rules_list: + external_address = rule.get('external_ip_address') + external_port = rule.get('external_port') or '' + local_address = rule.get('local_ip_address') + local_port = rule.get('local_port') or '' + protocol = protocol_map[rule.get('protocol', 0)] + options = ' '.join(decode_bitmask(rule.get('flags'))) + + values = [ + external_address, + external_port, + local_address, + local_port, + protocol, + options, + ] + data_entries.append(values) + headers = [ + 'External address', + 'External port', + 'Local address', + 'Local port', + 'Protocol', + 'Options', + ] + out = sorted(data_entries, key=lambda x: x[2]) + return tabulate(out, headers=headers, tablefmt='simple') + + +@_verify +def show_sessions(raw: bool): + vpp = VPPControl() + sessions_list: list[dict] = _get_raw_output_sessions(vpp.api) + + if raw: + return sessions_list + + else: + return _get_formatted_output_sessions(sessions_list) + + +@_verify +def show_summary(raw: bool): + vpp = VPPControl() + return vpp.cli_cmd('show nat44 summary').reply + + +@_verify +def show_static(raw: bool): + vpp = VPPControl() + nat_static_dump = vpp.api.nat44_static_mapping_dump() + rules_list: list[dict] = _get_raw_output(nat_static_dump) + + if raw: + return rules_list + + else: + return _get_formatted_output_rules(rules_list) + + +@_verify +def show_addresses(raw: bool): + vpp = VPPControl() + addresses_dump = vpp.api.nat44_address_dump() + addresses: list[dict] = _get_raw_output(addresses_dump) + + if raw: + return addresses + + else: + return _get_formatted_output_addresses(addresses) + + +@_verify +def show_interfaces(raw: bool): + vpp = VPPControl() + interfaces_dump = vpp.api.nat44_interface_dump() + interfaces: list[dict] = _get_raw_output(interfaces_dump) + + if raw: + return interfaces + + else: + return _get_formatted_output_interfaces(vpp, interfaces) + + +if __name__ == '__main__': + try: + res = vyos.opmode.run(sys.modules[__name__]) + if res: + print(res) + except (ValueError, vyos.opmode.Error) as e: + print(e) + sys.exit(1) -- cgit v1.2.3