summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorNataliia S. <81954790+natali-rs1985@users.noreply.github.com>2025-05-13 17:33:35 +0300
committerGitHub <noreply@github.com>2025-05-13 15:33:35 +0100
commit151afffe15ce89755f3c3f81a9d2c647e487f647 (patch)
treec823ad347680bc0644f073e136848bc2ce13002d /src
parent528248af9628c3170b4eac399f0bb339072d8eae (diff)
downloadvyos-1x-151afffe15ce89755f3c3f81a9d2c647e487f647.tar.gz
vyos-1x-151afffe15ce89755f3c3f81a9d2c647e487f647.zip
T7419: VPP ACL implementation (#31)
CLI: ``` set vpp acl ip tag-name <tag-name> rule <nn> action <permit|deny|permit-reflect> set vpp acl ip tag-name <tag-name> rule <nn> source prefix <prefix> set vpp acl ip tag-name <tag-name> rule <nn> source port <port|range> set vpp acl ip tag-name <tag-name> rule <nn> destination prefix <prefix> set vpp acl ip tag-name <tag-name> rule <nn> destination port <port> set vpp acl ip tag-name <tag-name> rule <nn> protocol <protocol> set vpp acl ip tag-name <tag-name> rule <nn> tcp-flags <fin|syn|rst|psh|ack|urg|ecn|cwr> set vpp acl ip tag-name <tag-name> rule <nn> tcp-flags not <fin|syn|rst|psh|ack|urg|ecn|cwr> set vpp acl ip interface <interface_name> input acl-tag <n> tag-name <tag-name> set vpp acl ip interface <interface_name> output acl-tag <n> tag-name <tag-name> set vpp acl macip tag-name <tag-name> rule <nn> prefix <prefix> set vpp acl macip tag-name <tag-name> rule <nn> mac-address <mac> set vpp acl macip tag-name <tag-name> rule <nn> mac-mask <mac-mask> set vpp acl macip tag-name <tag-name> rule <nn> action <permit|deny> set vpp acl macip interface <interface_name> tag-name <tag-name> ``` OP mode ``` show vpp acl ip tag-name <tag_name> show vpp acl ip interface show vpp acl macip tag-name <tag_name> show vpp acl macip interface ```
Diffstat (limited to 'src')
-rwxr-xr-xsrc/conf_mode/vpp.py4
-rw-r--r--src/conf_mode/vpp_acl.py393
-rw-r--r--src/conf_mode/vpp_interfaces_bonding.py10
-rw-r--r--src/conf_mode/vpp_interfaces_gre.py10
-rw-r--r--src/conf_mode/vpp_interfaces_ipip.py10
-rw-r--r--src/conf_mode/vpp_interfaces_loopback.py10
-rw-r--r--src/conf_mode/vpp_interfaces_vxlan.py10
-rw-r--r--src/op_mode/vpp_acl.py342
8 files changed, 789 insertions, 0 deletions
diff --git a/src/conf_mode/vpp.py b/src/conf_mode/vpp.py
index 3718f3897..ed0e80c2c 100755
--- a/src/conf_mode/vpp.py
+++ b/src/conf_mode/vpp.py
@@ -141,6 +141,10 @@ def get_config(config=None):
if conf.exists(['vpp', 'nat', 'cgnat']):
set_dependents('vpp_nat_cgnat', conf)
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
if not conf.exists(base):
return {
'removed_ifaces': removed_ifaces,
diff --git a/src/conf_mode/vpp_acl.py b/src/conf_mode/vpp_acl.py
new file mode 100644
index 000000000..28eae485b
--- /dev/null
+++ b/src/conf_mode/vpp_acl.py
@@ -0,0 +1,393 @@
+#!/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 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_protocol_by_name
+
+from vyos.vpp.utils import cli_ifaces_list
+from vyos.vpp.acl import Acl
+
+
+# TCP flag names to bit values
+TCP_FLAGS = {
+ 'FIN': 0x01,
+ 'SYN': 0x02,
+ 'RST': 0x04,
+ 'PSH': 0x08,
+ 'ACK': 0x10,
+ 'URG': 0x20,
+ 'ECN': 0x40,
+ 'CWR': 0x80,
+}
+
+# ACL action flags
+action_map = {
+ 'deny': 0,
+ 'permit': 1,
+ 'permit-reflect': 2,
+}
+
+
+def get_tcp_mask_value(set_flags, unset_flags):
+ mask = 0
+ value = 0
+
+ for flag in set_flags + unset_flags:
+ bit = TCP_FLAGS.get(flag.upper())
+ mask |= bit
+ if flag in set_flags:
+ value |= bit
+
+ return mask, value
+
+
+def get_port_first_last(port_range, protocol):
+ first_port = 0
+ last_port = 65535
+ if not port_range:
+ if protocol in ['icmp', 'ipv6-icmp']:
+ last_port = 255
+ elif '-' not in port_range:
+ first_port = last_port = port_range
+ else:
+ first_port, last_port = port_range.split('-')
+ return int(first_port), int(last_port)
+
+
+def create_ip_rules_list(rules):
+ rules_list = []
+ for rule in rules.values():
+ r = {
+ 'is_permit': action_map[rule.get('action')],
+ 'src_prefix': rule.get('source', {}).get('prefix', ''),
+ 'dst_prefix': rule.get('destination', {}).get('prefix', ''),
+ 'proto': (
+ int(get_protocol_by_name(rule.get('protocol')))
+ if rule.get('protocol') != 'all'
+ else 0
+ ),
+ }
+
+ tcp_flags = rule.get('tcp_flags', {})
+ set_flags = [flag for flag in tcp_flags if flag != 'not']
+ unet_flags = list(tcp_flags.get('not', {}).keys())
+ tcp_mask, tcp_value = get_tcp_mask_value(set_flags, unet_flags)
+ r['tcp_flags_mask'] = tcp_mask
+ r['tcp_flags_value'] = tcp_value
+
+ src_ports = rule.get('source', {}).get('port')
+ src_first_port, src_last_port = get_port_first_last(
+ src_ports, rule.get('protocol')
+ )
+ r['srcport_or_icmptype_first'] = src_first_port
+ r['srcport_or_icmptype_last'] = src_last_port
+
+ dst_ports = rule.get('destination', {}).get('port')
+ dst_first_port, dst_last_port = get_port_first_last(
+ dst_ports, rule.get('protocol')
+ )
+ r['dstport_or_icmpcode_first'] = dst_first_port
+ r['dstport_or_icmpcode_last'] = dst_last_port
+
+ rules_list.append(r)
+
+ return rules_list
+
+
+def create_macip_rules_list(rules):
+ rules_list = []
+ for rule in rules.values():
+ r = {
+ 'is_permit': action_map[rule.get('action')],
+ 'src_prefix': rule.get('prefix', ''),
+ 'src_mac': rule.get('mac_address', ''),
+ 'src_mac_mask': rule.get('mac_mask', ''),
+ }
+ rules_list.append(r)
+
+ return rules_list
+
+
+def get_config(config=None) -> dict:
+ if config:
+ conf = config
+ else:
+ conf = Config()
+
+ base = ['vpp', 'acl']
+
+ # 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,
+ )
+
+ # 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
+
+ changed_ip_ifaces = node_changed(
+ conf,
+ base + ['ip', 'interface'],
+ key_mangling=('-', '_'),
+ recursive=True,
+ expand_nodes=Diff.DELETE | Diff.ADD,
+ )
+
+ changed_macip_ifaces = node_changed(
+ conf,
+ base + ['macip', 'interface'],
+ key_mangling=('-', '_'),
+ recursive=True,
+ expand_nodes=Diff.DELETE | Diff.ADD,
+ )
+
+ config.update(
+ {
+ 'changed_ip_ifaces': changed_ip_ifaces,
+ 'changed_macip_ifaces': changed_macip_ifaces,
+ '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
+
+ for acl_type in ['ip', 'macip']:
+ if acl_type in config:
+ acl = config.get(acl_type)
+ if 'tag_name' not in acl:
+ raise ConfigError(f'"tag-name" is required for "acl {acl_type}"')
+
+ for acl_name, acl_config in acl.get('tag_name').items():
+ if 'rule' not in acl_config:
+ raise ConfigError(f'Rules must be configured for ACL {acl_name}')
+
+ for rule, rule_config in acl_config.get('rule').items():
+ err_msg = f'Configuration error for {acl_type} ACL {acl_name} in rule {rule}:'
+ if 'action' not in rule_config:
+ raise ConfigError(f'{err_msg} action must be defined')
+
+ for iface, iface_config in acl.get('interface', {}).items():
+ if iface not in config.get('vpp_ifaces'):
+ raise ConfigError(
+ f'{iface} must be a VPP interface for ACL interface'
+ )
+
+ if 'ip' in config:
+ acl = config.get('ip')
+ for acl_name, acl_config in acl.get('tag_name').items():
+ for rule, rule_config in acl_config.get('rule').items():
+ err_msg = (
+ f'Configuration error for {acl_type} ACL {acl_name} in rule {rule}:'
+ )
+
+ # verify IPv4 and IPv6 address family
+ src_prefix = rule_config.get('source', {}).get('prefix')
+ dst_prefix = rule_config.get('destination', {}).get('prefix')
+ src = ipaddress.ip_network(src_prefix) if src_prefix else None
+ dst = ipaddress.ip_network(dst_prefix) if dst_prefix else None
+
+ if src and dst:
+ if src.version != dst.version:
+ raise ConfigError(
+ f'{err_msg} source and destination prefixes must be from the same IP family'
+ )
+ elif src or dst:
+ family = src.version if src else dst.version
+ if family == 6:
+ raise ConfigError(
+ f'{err_msg} both source and destination prefixes must be defined for IPv6'
+ )
+
+ # verify protocol
+ protocol = rule_config.get('protocol')
+ if protocol != 'all':
+ proto = get_protocol_by_name(protocol)
+ if not isinstance(proto, int) and (
+ not proto.isdigit() or int(proto) > 147
+ ):
+ raise ConfigError(
+ f'{err_msg} protocol name {protocol} is not valid'
+ )
+
+ # verify TCP flags
+ if 'tcp_flags' in rule_config:
+ if rule_config.get('protocol') != 'tcp':
+ raise ConfigError(
+ f'{err_msg} protocol must be tcp when specifying tcp flags'
+ )
+
+ not_flags = rule_config.get('tcp_flags').get('not', [])
+ if not_flags:
+ duplicates = [
+ flag
+ for flag in rule_config.get('tcp_flags')
+ if flag in not_flags
+ ]
+ if duplicates:
+ raise ConfigError(
+ f'{err_msg} cannot match a tcp flag as set and not set: {duplicates}'
+ )
+
+ for iface, iface_config in acl.get('interface', {}).items():
+ if not any(key in iface_config for key in ('input', 'output')):
+ raise ConfigError(
+ f'Please specify direction input/output for interface {iface}'
+ )
+
+ for direction in ['input', 'output']:
+ if direction in iface_config:
+ iface_acl = iface_config.get(direction)
+ if 'acl_tag' not in iface_acl:
+ raise ConfigError(
+ f'"acl-tag" is required for {direction} interface {iface}'
+ )
+
+ used_names = []
+ for tag, tag_conf in iface_acl.get('acl_tag').items():
+ if 'tag_name' not in tag_conf:
+ raise ConfigError(
+ f'"tag-name" is required for {direction} interface {iface} with acl-tag {tag}'
+ )
+ name = tag_conf.get('tag_name')
+ if name not in acl.get('tag_name').keys():
+ raise ConfigError(
+ f'ACL with tag-name {name} does not exist. '
+ f'Cannot use it for {direction} interface {iface}'
+ )
+ if name in used_names:
+ raise ConfigError(
+ f'ACL with tag-name {name} is already used for {direction} interface {iface}'
+ )
+ used_names.append(name)
+
+ if 'macip' in config:
+ acl = config.get('macip')
+ for iface, iface_config in acl.get('interface', {}).items():
+ if 'tag_name' not in iface_config:
+ raise ConfigError(f'"tag-name" is required for interface {iface}')
+ name = iface_config.get('tag_name')
+ if name not in acl.get('tag_name').keys():
+ raise ConfigError(
+ f'ACL with tag-name {name} does not exist. Cannot use it for interface {iface}'
+ )
+
+
+def generate(config):
+ pass
+
+
+def apply(config):
+ acl = Acl()
+
+ if 'effective' in config:
+ # Delete ACL ip
+ if 'ip' in config.get('effective'):
+ remove_config_ip = config.get('effective').get('ip')
+
+ # Delete ACL interfaces
+ for interface in config.get('changed_ip_ifaces'):
+ acl.delete_acl_interface(interface)
+
+ # Delete ACLs
+ for acl_name in remove_config_ip.get('tag_name'):
+ if acl_name not in config.get('ip', {}).get('tag_name', {}):
+ acl.delete_acl(acl_name)
+
+ # Delete ACL macip
+ if 'macip' in config.get('effective'):
+ remove_config_macip = config.get('effective').get('macip')
+
+ # Delete ACL interfaces
+ for interface in config.get('changed_macip_ifaces'):
+ acl.delete_acl_macip_interface(interface)
+
+ # Delete ACL macip
+ for acl_name in remove_config_macip.get('tag_name'):
+ if acl_name not in config.get('macip', {}).get('tag_name', {}):
+ acl.delete_acl_macip(acl_name)
+
+ if 'remove' in config:
+ return None
+
+ # Add or replace ACL ip
+ config_ip = config.get('ip', {})
+ for acl_name in config_ip.get('tag_name', {}):
+ rules = create_ip_rules_list(
+ config_ip.get('tag_name').get(acl_name).get('rule')
+ )
+ acl.add_replace_acl(acl_name, rules)
+
+ for iface, iface_config in config_ip.get('interface', {}).items():
+ input_tags = [
+ v['tag_name']
+ for v in iface_config.get('input', {}).get('acl_tag', {}).values()
+ ]
+ output_tags = [
+ v['tag_name']
+ for v in iface_config.get('output', {}).get('acl_tag', {}).values()
+ ]
+ acl.add_acl_interface(iface, input_tags, output_tags)
+
+ # Add or replace ACL macip
+ config_macip = config.get('macip', {})
+ for acl_name in config_macip.get('tag_name', {}):
+ rules = create_macip_rules_list(
+ config_macip.get('tag_name').get(acl_name).get('rule')
+ )
+ acl.add_replace_acl_macip(acl_name, rules)
+
+ for iface, iface_config in config_macip.get('interface', {}).items():
+ acl.add_acl_macip_interface(iface, iface_config.get('tag_name'))
+
+
+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_interfaces_bonding.py b/src/conf_mode/vpp_interfaces_bonding.py
index 7ab1abb24..2b17acb24 100644
--- a/src/conf_mode/vpp_interfaces_bonding.py
+++ b/src/conf_mode/vpp_interfaces_bonding.py
@@ -139,6 +139,16 @@ def get_config(config=None) -> dict:
if conf.exists(['vpp', 'kernel-interfaces', iface]):
set_dependents('vpp_kernel_interface', conf, iface)
+ # NAT dependency
+ if conf.exists(['vpp', 'nat44']):
+ set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
+
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
config['ifname'] = ifname
return config
diff --git a/src/conf_mode/vpp_interfaces_gre.py b/src/conf_mode/vpp_interfaces_gre.py
index 823f3ff4d..f6da2d55d 100644
--- a/src/conf_mode/vpp_interfaces_gre.py
+++ b/src/conf_mode/vpp_interfaces_gre.py
@@ -110,6 +110,16 @@ def get_config(config=None) -> dict:
if conf.exists(['vpp', 'kernel-interfaces', iface]):
set_dependents('vpp_kernel_interface', conf, iface)
+ # NAT dependency
+ if conf.exists(['vpp', 'nat44']):
+ set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
+
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
config['ifname'] = ifname
return config
diff --git a/src/conf_mode/vpp_interfaces_ipip.py b/src/conf_mode/vpp_interfaces_ipip.py
index c52e7c997..f40a4e243 100644
--- a/src/conf_mode/vpp_interfaces_ipip.py
+++ b/src/conf_mode/vpp_interfaces_ipip.py
@@ -109,6 +109,16 @@ def get_config(config=None) -> dict:
if iface in config.get('vpp_kernel_interfaces'):
set_dependents('vpp_kernel_interface', conf, iface)
+ # NAT dependency
+ if conf.exists(['vpp', 'nat44']):
+ set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
+
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
config['ifname'] = ifname
return config
diff --git a/src/conf_mode/vpp_interfaces_loopback.py b/src/conf_mode/vpp_interfaces_loopback.py
index f0bcdfb18..2b37b5e1c 100644
--- a/src/conf_mode/vpp_interfaces_loopback.py
+++ b/src/conf_mode/vpp_interfaces_loopback.py
@@ -99,6 +99,16 @@ def get_config(config=None) -> dict:
if conf.exists(['vpp', 'kernel-interfaces', iface]):
set_dependents('vpp_kernel_interface', conf, iface)
+ # NAT dependency
+ if conf.exists(['vpp', 'nat44']):
+ set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
+
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
config['ifname'] = ifname
return config
diff --git a/src/conf_mode/vpp_interfaces_vxlan.py b/src/conf_mode/vpp_interfaces_vxlan.py
index 963cd90ec..065a8eafb 100644
--- a/src/conf_mode/vpp_interfaces_vxlan.py
+++ b/src/conf_mode/vpp_interfaces_vxlan.py
@@ -116,6 +116,16 @@ def get_config(config=None) -> dict:
if conf.exists(['vpp', 'kernel-interfaces', iface]):
set_dependents('vpp_kernel_interface', conf, iface)
+ # NAT dependency
+ if conf.exists(['vpp', 'nat44']):
+ set_dependents('vpp_nat', conf)
+ if conf.exists(['vpp', 'nat', 'cgnat']):
+ set_dependents('vpp_nat_cgnat', conf)
+
+ # ACL dependency
+ if conf.exists(['vpp', 'acl']):
+ set_dependents('vpp_acl', conf)
+
config['ifname'] = ifname
return config
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)