diff options
Diffstat (limited to 'src/op_mode')
| -rwxr-xr-x | src/op_mode/show_vpp_interfaces.py | 257 | ||||
| -rw-r--r-- | src/op_mode/show_vpp_nat44.py | 251 | ||||
| -rw-r--r-- | src/op_mode/vpp_acl.py | 342 | ||||
| -rw-r--r-- | src/op_mode/vpp_nat_cgnat.py | 140 |
4 files changed, 990 insertions, 0 deletions
diff --git a/src/op_mode/show_vpp_interfaces.py b/src/op_mode/show_vpp_interfaces.py new file mode 100755 index 000000000..b7c42eb58 --- /dev/null +++ b/src/op_mode/show_vpp_interfaces.py @@ -0,0 +1,257 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2023-2024 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 argparse +import json +from tabulate import tabulate + +from vyos.configquery import ConfigTreeQuery +from vyos.utils.process import rc_cmd + +from vyos.vpp import VPPControl +from vyos.vpp.utils import ( + vpp_ifaces_list, + vpp_ip_addresses_by_index, + vpp_ifaces_stats, +) + + +def get_iproute_address_list(interface: str) -> list: + """Get data from the Linux command 'ip --json address list dev {interface}' and return a list info + for the given interface. + + Args: + interface (str): Interface name. + + Returns: + list: A dictionary containing the JSON data from the 'ip --json address list' command for the specified interface. + """ + rc, out = rc_cmd(f'ip --json address list dev {interface}') + if rc: + return [] + return json.loads(out) + + +def get_iproute_link_list(interface): + """Get data from the Linux command 'ip --json link show dev {interface}' and return a list info + for the given interface. + + Args: + interface (str): Interface name. + + Returns: + list: A dictionary containing the JSON data from the 'ip --json link show' command for the specified interface. + """ + rc, out = rc_cmd(f'ip --json link list dev {interface}') + if rc: + return [] + return json.loads(out) + + +def merge_dicts(*dicts) -> dict: + """Merge dictionaries into a new dictionary. + + Args: + *dicts: Any number of dictionaries. + + Returns: + dict: A new dictionary containing all the key-value pairs from the given dictionaries. + """ + merged = {} + for dictionary in dicts: + merged.update(dictionary) + return merged + + +def show_interfaces(interfaces_list: list) -> str: + """Get JSON info from linux and represent it in a table format + Use tabulate to generate table + + Interface IP Address Mtu S/L Description + --------- ---------- --- --- ----------- + dum0 203.0.113.1/32 1500 u/u + 100.64.1.1/24 + eth0 192.168.122.14/24 1500 u/u WAN + + :return: + """ + table = [] + for interface in interfaces_list: + # Get the data for the interface + ip_address_data = get_iproute_address_list(interface) + link_data = get_iproute_link_list(interface) + + # Skip this interface if data is not available + if not link_data: + continue + interface_data = merge_dicts(ip_address_data[0], link_data[0]) + + # Get the interface name + interface_name = interface_data['ifname'] + + # Get the IP addresses and their corresponding prefixes + ip_info = [ + (address['local'], address.get('prefixlen', '')) + for address in interface_data['addr_info'] + ] + + # Format the IP addresses with prefixes and line breaks + ip_addresses = '\n'.join(f'{ip}/{prefix}' for ip, prefix in ip_info) + + # Get the MAC address + mac = interface_data.get('address', 'n/a') + + # Get the MTU + mtu = interface_data.get('mtu') + + # Get the state of the interface + state = interface_data['operstate'].lower() + + # Get the description of the interface + description = interface_data.get('ifalias', '') + + # Create the list of values for the table + values = [interface_name, ip_addresses, mac, mtu, state, description] + + # Append the list of values to the table + table.append(values) + + # Print the table with IP addresses listed on separate lines + headers = ['Interface', 'IP Address', 'MAC', 'MTU', 'State', 'Description'] + return tabulate(table, headers=headers, tablefmt='simple') + + +def show_interfaces_dataplane(interfaces_list: list, filter_type: str = 'all') -> str: + table = [] + interface_dp_filter = ('tun', 'tap') + lcp_pair_list = vpp.lcp_pairs_list() + vpp_name_kernel_to_kernel_name = { + entry['vpp_name_kernel']: entry['kernel_name'] for entry in lcp_pair_list + } + for interface in interfaces_list: + interface_name = interface.get('interface_name') + if filter_type == 'no_tun_tap' and interface_name.startswith( + interface_dp_filter + ): + continue + if filter_type == 'only_tun_tap' and not interface_name.startswith( + interface_dp_filter + ): + continue + kernel_name = vpp_name_kernel_to_kernel_name.get(interface_name, '') + + dp_ip_addresses = vpp_ip_addresses_by_index( + vpp.api, interface.get('sw_if_index') + ) + ip_addresses = '\n'.join(dp_ip_addresses) + + mac = str(interface.get('l2_address', 'n/a')) + mtu = interface.get('mtu', [])[0] + # state + flags = interface.get('flags') + state = 'up' if flags == 3 else 'down' + + iftype = interface.get('interface_dev_type').split()[0] + + values = [kernel_name, interface_name, iftype, ip_addresses, mac, mtu, state] + table.append(values) + headers = [ + 'Kernel', + 'Dataplane', + 'Type', + 'IP Address', + 'MAC', + 'MTU', + 'State', + ] + table = sorted(table) + return tabulate(table, headers=headers, tablefmt='simple') + + +def show_interfaces_hardware(intf_name) -> str: + if not intf_name: + intf_name = '' + + statistics = vpp_ifaces_stats(intf_name) + for intf, stats in sorted(statistics.items()): + print(f'\n---------------------------------\nInterface {intf}:\n') + table = [] + for k, v in stats.items(): + if isinstance(v, dict): + for i, j in v.items(): + table.append([f"{k} {i}", j]) + else: + table.append([k, v]) + print(tabulate(table, tablefmt="presto")) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='Show VPP interfaces') + parser.add_argument( + '--summary', + action='store_true', + help='Show summary of VPP interfaces (ethernet and kernel tun)', + ) + parser.add_argument( + '--dataplane', action='store_true', help='Show VPP ethernet interfaces' + ) + parser.add_argument( + '--kernel', action='store_true', help='Show VPP kernel interfaces' + ) + parser.add_argument( + '--iproute', action='store_true', help='Show interfaces (iproute2)' + ) + parser.add_argument( + '--hardware', + action='store_true', + help='Show more detailed statistics for VPP interfaces', + ) + parser.add_argument('--intf-name', action='store', help='Kernel interface name') + + args = parser.parse_args() + + config = ConfigTreeQuery() + + if not config.exists('vpp settings interface'): + print('VPP interfaces not configured') + exit(0) + + vpp = VPPControl() + dp_ifaces_list = vpp_ifaces_list(vpp.api) + + if args.summary: + print(show_interfaces_dataplane(dp_ifaces_list, filter_type='all')) + + if args.dataplane: + print(show_interfaces_dataplane(dp_ifaces_list, filter_type='no_tun_tap')) + exit(0) + + if args.kernel: + print(show_interfaces_dataplane(dp_ifaces_list, filter_type='only_tun_tap')) + + if args.iproute: + vpp_interfaces = [] + vpp_ethernet = config.list_nodes('vpp settings interface') + vpp_interfaces.extend(vpp_ethernet) + if config.exists('vpp kernel-interfaces'): + vpp_kernel_interfaces = config.list_nodes('vpp kernel-interfaces') + vpp_interfaces.extend(vpp_kernel_interfaces) + print(show_interfaces(interfaces_list=vpp_interfaces)) + + if args.hardware: + show_interfaces_hardware(intf_name=args.intf_name) diff --git a/src/op_mode/show_vpp_nat44.py b/src/op_mode/show_vpp_nat44.py new file mode 100644 index 000000000..d0569ac49 --- /dev/null +++ b/src/op_mode/show_vpp_nat44.py @@ -0,0 +1,251 @@ +#!/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. + +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_acl.py b/src/op_mode/vpp_acl.py new file mode 100644 index 000000000..7afe96433 --- /dev/null +++ b/src/op_mode/vpp_acl.py @@ -0,0 +1,342 @@ +#!/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. + +import json +import sys +import typing +from tabulate import tabulate + +import vyos.opmode +from vyos.config import Config +from vyos.configquery import ConfigTreeQuery + +from vyos.vpp import VPPControl + + +NO_ACL_INDEX = 0xFFFFFFFF + +# ACL action flags +action_map = { + 0: 'deny', + 1: 'permit', + 2: 'permit-reflect', +} + +# TCP flag names to bit values +TCP_FLAGS = { + 'FIN': 0x01, + 'SYN': 0x02, + 'RST': 0x04, + 'PSH': 0x08, + 'ACK': 0x10, + 'URG': 0x20, + 'ECN': 0x40, + 'CWR': 0x80, +} + + +def _verify(target): + """Decorator checks if config for VPP NAT CGNAT exists""" + from functools import wraps + + if target not in ['ip', 'macip', 'no_target']: + raise ValueError('Invalid target') + + def _verify_target(func): + @wraps(func) + def _wrapper(*args, **kwargs): + config = ConfigTreeQuery() + path = 'vpp acl' + if target == 'ip': + path += ' ip' + elif target == 'macip': + path += ' macip' + if not config.exists(path): + raise vyos.opmode.UnconfiguredSubsystem(f'"{path}" is not configured') + return func(*args, **kwargs) + + return _wrapper + + return _verify_target + + +def _get_acl_tag_by_index(vpp, acl_index): + acl = vpp.api.acl_dump(acl_index=acl_index) + if acl: + return acl[0].tag + + return None + + +def _get_macip_acl_tag_by_index(vpp, acl_index): + acl = vpp.api.macip_acl_dump(acl_index=acl_index) + if acl: + return acl[0].tag + + return None + + +def _get_tcp_flag_states(value, mask): + set_flags = [] + unset_flags = [] + for flag, bit in TCP_FLAGS.items(): + if mask & bit: # This flag is being checked + if value & bit: + set_flags.append(flag) + else: + unset_flags.append(flag) + return sorted(set_flags), sorted(unset_flags) + + +def _get_raw_output_acls(data_dump): + out = [] + for data in data_dump: + rules = [json.loads(json.dumps(d._asdict(), default=str)) for d in data.r] + out.append( + { + 'acl_index': data.acl_index, + 'tag': data.tag, + 'count': data.count, + 'r': rules, + } + ) + return out + + +def _get_raw_output_interfaces(data_dump): + ifaces_list = [] + for iface in data_dump: + if iface.count != 0: + ifaces_list.append(json.loads(json.dumps(iface._asdict(), default=str))) + return ifaces_list + + +def _get_formatted_output_interfaces(vpp, interfaces): + data_entries = [] + for interface in interfaces: + name = vpp.get_interface_name(interface.get('sw_if_index')) + input_acls = [] + for acl_index in interface.get('acls')[: interface.get('n_input')]: + input_acls.append(_get_acl_tag_by_index(vpp, int(acl_index))) + output_acls = [] + for acl_index in interface.get('acls')[interface.get('n_input') :]: + output_acls.append(_get_acl_tag_by_index(vpp, int(acl_index))) + values = [ + name, + '\n'.join(input_acls), + '\n'.join(output_acls), + ] + data_entries.append(values) + + headers = ['Interface', 'Input ACLs', 'Output ACLs'] + return tabulate(data_entries, headers=headers, tablefmt='simple') + + +def _get_formatted_output_macip_interfaces(vpp, interfaces): + data_entries = [] + for interface in interfaces: + name = vpp.get_interface_name(interface.get('sw_if_index')) + acl = _get_macip_acl_tag_by_index(vpp, int(interface.get('acls')[0])) + data_entries.append([name, acl]) + + headers = ['Interface', 'ACL'] + return tabulate(data_entries, headers=headers, tablefmt='simple') + + +def _get_formatted_output_acls(acls_list): + conf = Config() + + for acl in acls_list: + acl_index = acl.get('acl_index') + tag = acl.get('tag') + rules = acl.get('r') + print( + '\n---------------------------------\n' + f'IP ACL "tag-name {tag}" acl_index {acl_index}\n' + ) + + path = ['vpp', 'acl', 'ip', 'tag-name', tag, 'rule'] + conf_rules = conf.list_nodes(path) + data_entries = [] + for rule_index, rule in enumerate(rules): + srcport_first = str(rule.get('srcport_or_icmptype_first')) + srcport_last = str(rule.get('srcport_or_icmptype_last')) + dstport_first = str(rule.get('dstport_or_icmpcode_first')) + dstport_last = str(rule.get('dstport_or_icmpcode_last')) + set_flags, unset_flags = _get_tcp_flag_states( + rule.get('tcp_flags_value'), rule.get('tcp_flags_mask') + ) + + values = [ + conf_rules[rule_index], + action_map.get(rule.get('is_permit')), + rule.get('src_prefix'), + ( + f'{srcport_first}-{srcport_last}' + if srcport_first != srcport_last + else srcport_first + ), + rule.get('dst_prefix'), + ( + f'{dstport_first}-{dstport_last}' + if dstport_first != dstport_last + else dstport_first + ), + rule.get('proto'), + '\n'.join(set_flags), + '\n'.join(unset_flags), + ] + data_entries.append(values) + + headers = [ + 'Rule', + 'Action', + 'Src prefix', + 'Src port', + 'Dst prefix', + 'Dst port', + 'Proto', + 'TCP flags set', + 'TCP flags not set', + ] + print(tabulate(data_entries, headers=headers, tablefmt='simple')) + print('\n') + + +def _get_formatted_output_macip_acls(acls_list): + conf = Config() + + for acl in acls_list: + acl_index = acl.get('acl_index') + tag = acl.get('tag') + rules = acl.get('r') + print( + '\n---------------------------------\n' + f'MACIP ACL "tag-name {tag}" acl_index {acl_index}\n' + ) + + path = ['vpp', 'acl', 'macip', 'tag-name', tag, 'rule'] + conf_rules = conf.list_nodes(path) + data_entries = [] + for rule_index, rule in enumerate(rules): + values = [ + conf_rules[rule_index], + action_map.get(rule.get('is_permit')), + rule.get('src_prefix'), + rule.get('src_mac'), + rule.get('src_mac_mask'), + ] + data_entries.append(values) + + headers = [ + 'Rule', + 'Action', + 'IP prefix', + 'MAC address', + 'MAC mask', + ] + print(tabulate(data_entries, headers=headers, tablefmt='simple')) + print('\n') + + +def _find_acl_by_tag(acls, tag_name): + return [acl for acl in acls if acl['tag'] == tag_name] + + +@_verify('ip') +def show_ip_acls(raw: bool, tag_name: typing.Optional[str]): + vpp = VPPControl() + acls_dump = vpp.api.acl_dump(acl_index=NO_ACL_INDEX) + acls: list[dict] = _get_raw_output_acls(acls_dump) + + if tag_name: + acls = _find_acl_by_tag(acls, tag_name) + + if raw: + return acls + + else: + return _get_formatted_output_acls(acls) + + +@_verify('macip') +def show_macip_acls(raw: bool, tag_name: typing.Optional[str]): + vpp = VPPControl() + acls_dump = vpp.api.macip_acl_dump(acl_index=NO_ACL_INDEX) + acls: list[dict] = _get_raw_output_acls(acls_dump) + + if tag_name: + acls = _find_acl_by_tag(acls, tag_name) + + if raw: + return acls + + else: + return _get_formatted_output_macip_acls(acls) + + +@_verify('ip') +def show_interfaces(raw: bool): + vpp = VPPControl() + interfaces_dump = vpp.api.acl_interface_list_dump() + interfaces: list[dict] = _get_raw_output_interfaces(interfaces_dump) + + if raw: + return interfaces + + else: + return _get_formatted_output_interfaces(vpp, interfaces) + + +@_verify('macip') +def show_macip_interfaces(raw: bool): + vpp = VPPControl() + interfaces_dump = vpp.api.macip_acl_interface_list_dump() + interfaces: list[dict] = _get_raw_output_interfaces(interfaces_dump) + + if raw: + return interfaces + + else: + return _get_formatted_output_macip_interfaces(vpp, interfaces) + + +@_verify('no_target') +def show_all_acls(raw: bool): + conf = Config() + acls_all = {} + path = ['vpp', 'acl'] + if conf.exists(path + ['ip']): + ip_acls = show_ip_acls(raw, tag_name=None) + acls_all['ip'] = ip_acls + if conf.exists(path + ['macip']): + macip_acls = show_macip_acls(raw, tag_name=None) + acls_all['macip'] = macip_acls + + if raw: + return acls_all + + +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_cgnat.py b/src/op_mode/vpp_nat_cgnat.py new file mode 100644 index 000000000..5112c70be --- /dev/null +++ b/src/op_mode/vpp_nat_cgnat.py @@ -0,0 +1,140 @@ +#!/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. + +import json +import sys +from tabulate import tabulate + +import vyos.opmode +from vyos.configquery import ConfigTreeQuery + +from vyos.vpp import VPPControl + + +def _verify(func): + """Decorator checks if config for VPP NAT CGNAT exists""" + from functools import wraps + + @wraps(func) + def _wrapper(*args, **kwargs): + config = ConfigTreeQuery() + base = 'vpp nat cgnat' + if not config.exists(base): + raise vyos.opmode.UnconfiguredSubsystem(f'{base} is not configured') + + return func(*args, **kwargs) + + return _wrapper + + +def _get_raw_output(data_dump): + data = [json.loads(json.dumps(d._asdict(), default=str)) for d in data_dump] + return data + + +def _get_formatted_output_interfaces(vpp, interfaces): + print('CGNAT interfaces:') + for interface in interfaces: + name = vpp.get_interface_name(interface['sw_if_index']) + iface_type = 'in' if interface['is_inside'] else 'out' + print(f' {name} {iface_type}') + + +def _get_formatted_output_mappings(rules_list): + data_entries = [] + for rule in rules_list: + in_addr = rule.get('in_addr') + in_plen = str(rule.get('in_plen')) + out_addr = rule.get('out_addr') + out_plen = str(rule.get('out_plen')) + sharing_ratio = rule.get('sharing_ratio') + ports_per_host = rule.get('ports_per_host') + ses_num = rule.get('ses_num') + + values = [ + f'{in_addr}/{in_plen}', + f'{out_addr}/{out_plen}', + sharing_ratio, + ports_per_host, + ses_num, + ] + data_entries.append(values) + headers = [ + 'Inside', + 'Outside', + 'Sharing ratio', + 'Ports per host', + 'Sessions', + ] + out = sorted(data_entries, key=lambda x: x[0]) + return tabulate(out, headers=headers, tablefmt='simple') + + +@_verify +def show_sessions(raw: bool): + vpp = VPPControl() + out = vpp.cli_cmd('show det44 sessions').reply + out = out.replace('NAT44 deterministic', 'CGNAT') + return out + + +@_verify +def show_mappings(raw: bool): + vpp = VPPControl() + nat_static_dump = vpp.api.det44_map_dump() + rules_list: list[dict] = _get_raw_output(nat_static_dump) + + if raw: + return rules_list + + else: + return _get_formatted_output_mappings(rules_list) + + +@_verify +def show_interfaces(raw: bool): + vpp = VPPControl() + interfaces_dump = vpp.api.det44_interface_dump() + interfaces: list[dict] = _get_raw_output(interfaces_dump) + + if raw: + return interfaces + + else: + return _get_formatted_output_interfaces(vpp, interfaces) + + +@_verify +def clear_session(address: str, port: str, ext_address: str, ext_port: str): + vpp = VPPControl() + vpp.api.det44_close_session_in( + in_addr=address, + in_port=int(port), + ext_addr=ext_address, + ext_port=int(ext_port), + ) + + +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) |