#!/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_protocol_by_name from vyos.vpp.utils import cli_ifaces_list from vyos.vpp.utils import vpp_iface_name_transform from vyos.vpp.acl import Acl from vyos.vpp.config_verify import verify_vpp_interface_not_a_member # 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 = tcp_flags.get('is_set', []) unet_flags = tcp_flags.get('is_not_set', []) 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_mac_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, ) 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 changed_ip_ifaces = node_changed( conf, base + ['ip', 'interface'], key_mangling=('-', '_'), recursive=True, expand_nodes=Diff.DELETE | Diff.ADD, ) changed_mac_ifaces = node_changed( conf, base + ['mac', 'interface'], key_mangling=('-', '_'), recursive=True, expand_nodes=Diff.DELETE | Diff.ADD, ) config.update( { 'changed_ip_ifaces': changed_ip_ifaces, 'changed_mac_ifaces': changed_mac_ifaces, 'vpp_ifaces': cli_ifaces_list(conf, include_vifs=True), } ) if effective_config: config.update({'effective': effective_config}) # VPP interface membership data for member-conflict checks config['interfaces_vpp'] = conf.get_config_dict( ['interfaces', 'vpp'], key_mangling=('-', '_'), get_first_key=True, no_tag_node_value_mangle=True, ) return config def verify(config): if 'remove' in config or 'remove_vpp' in config: return None for acl_type in ['ip', 'mac']: 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') not_allowed_prefixes = ('vppbr', 'vppxcon') for iface, iface_config in acl.get('interface', {}).items(): if iface.startswith(not_allowed_prefixes): raise ConfigError( f'Interface {iface} is not allowed in ACL {acl_type} configuration' ) if iface not in config.get('vpp_ifaces'): raise ConfigError( f'{iface} must be a VPP interface for ACL interface' ) verify_vpp_interface_not_a_member(iface, config) 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' ) tcp_flags = rule_config.get('tcp_flags', {}) flags_set = tcp_flags.get('is_set', []) flags_not_set = tcp_flags.get('is_not_set', []) # same flag cannot be both set and not set conflict = [flag for flag in flags_set if flag in flags_not_set] if conflict: raise ConfigError( f'{err_msg} cannot match a TCP flag as both set and not set: ' f'{", ".join(sorted(conflict))}' ) 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 'mac' in config: acl = config.get('mac') 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}' ) return None def generate(config): pass def apply(config): if 'remove_vpp' in config: return None 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(vpp_iface_name_transform(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 mac if 'mac' in config.get('effective'): remove_config_mac = config.get('effective').get('mac') # Delete ACL interfaces for interface in config.get('changed_mac_ifaces'): acl.delete_acl_mac_interface(vpp_iface_name_transform(interface)) # Delete ACL mac for acl_name in remove_config_mac.get('tag_name'): if acl_name not in config.get('mac', {}).get('tag_name', {}): acl.delete_acl_mac(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(vpp_iface_name_transform(iface), input_tags, output_tags) # Add or replace ACL mac config_mac = config.get('mac', {}) for acl_name in config_mac.get('tag_name', {}): rules = create_mac_rules_list( config_mac.get('tag_name').get(acl_name).get('rule') ) acl.add_replace_acl_mac(acl_name, rules) for iface, iface_config in config_mac.get('interface', {}).items(): acl.add_acl_mac_interface( vpp_iface_name_transform(iface), iface_config.get('tag_name'), ) return None if __name__ == '__main__': try: c = get_config() verify(c) generate(c) apply(c) except ConfigError as e: print(e) exit(1)