diff options
Diffstat (limited to 'src')
| -rwxr-xr-x | src/conf_mode/firewall.py | 11 | ||||
| -rwxr-xr-x | src/conf_mode/interfaces_wireguard.py | 28 | ||||
| -rwxr-xr-x | src/conf_mode/nat.py | 6 | ||||
| -rwxr-xr-x | src/conf_mode/policy.py | 16 | ||||
| -rwxr-xr-x | src/conf_mode/service_monitoring_prometheus.py | 18 | ||||
| -rwxr-xr-x | src/conf_mode/service_pppoe-server.py | 4 | ||||
| -rwxr-xr-x | src/conf_mode/vpn_ipsec.py | 2 | ||||
| -rwxr-xr-x | src/conf_mode/vpn_openconnect.py | 2 | ||||
| -rwxr-xr-x | src/etc/opennhrp/opennhrp-script.py | 371 | ||||
| -rw-r--r-- | src/systemd/opennhrp.service | 13 | ||||
| -rwxr-xr-x | src/validators/bgp-large-community-list | 21 |
11 files changed, 72 insertions, 420 deletions
diff --git a/src/conf_mode/firewall.py b/src/conf_mode/firewall.py index 274ca2ce6..348eaeba3 100755 --- a/src/conf_mode/firewall.py +++ b/src/conf_mode/firewall.py @@ -17,6 +17,8 @@ import os import re +from glob import glob + from sys import exit from vyos.base import Warning from vyos.config import Config @@ -30,6 +32,7 @@ from vyos.firewall import geoip_update from vyos.template import render from vyos.utils.dict import dict_search_args from vyos.utils.dict import dict_search_recursive +from vyos.utils.file import write_file from vyos.utils.process import call from vyos.utils.process import cmd from vyos.utils.process import rc_cmd @@ -37,7 +40,6 @@ from vyos.utils.network import get_vrf_members from vyos.utils.network import get_interface_vrf from vyos import ConfigError from vyos import airbag -from pathlib import Path from subprocess import run as subp_run airbag.enable() @@ -626,10 +628,11 @@ def apply(firewall): domain_action = 'restart' if dict_search_args(firewall, 'group', 'remote_group') or dict_search_args(firewall, 'group', 'domain_group') or firewall['ip_fqdn'].items() or firewall['ip6_fqdn'].items(): text = f'# Automatically generated by firewall.py\nThis file indicates that vyos-domain-resolver service is used by the firewall.\n' - Path(domain_resolver_usage).write_text(text) + write_file(domain_resolver_usage, text) else: - Path(domain_resolver_usage).unlink(missing_ok=True) - if not Path('/run').glob('use-vyos-domain-resolver*'): + if os.path.exists(domain_resolver_usage): + os.unlink(domain_resolver_usage) + if not glob('/run/use-vyos-domain-resolver*'): domain_action = 'stop' call(f'systemctl {domain_action} vyos-domain-resolver.service') diff --git a/src/conf_mode/interfaces_wireguard.py b/src/conf_mode/interfaces_wireguard.py index 3ca6ecdca..770667df1 100755 --- a/src/conf_mode/interfaces_wireguard.py +++ b/src/conf_mode/interfaces_wireguard.py @@ -14,6 +14,9 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. +import os + +from glob import glob from sys import exit from vyos.config import Config @@ -35,7 +38,6 @@ from vyos.utils.network import is_wireguard_key_pair from vyos.utils.process import call from vyos import ConfigError from vyos import airbag -from pathlib import Path airbag.enable() @@ -145,19 +147,11 @@ def generate(wireguard): def apply(wireguard): check_kmod('wireguard') - if 'rebuild_required' in wireguard or 'deleted' in wireguard: - wg = WireGuardIf(**wireguard) - # WireGuard only supports peer removal based on the configured public-key, - # by deleting the entire interface this is the shortcut instead of parsing - # out all peers and removing them one by one. - # - # Peer reconfiguration will always come with a short downtime while the - # WireGuard interface is recreated (see below) - wg.remove() + wg = WireGuardIf(**wireguard) - # Create the new interface if required - if 'deleted' not in wireguard: - wg = WireGuardIf(**wireguard) + if 'deleted' in wireguard: + wg.remove() + else: wg.update(wireguard) domain_resolver_usage = '/run/use-vyos-domain-resolver-interfaces-wireguard-' + wireguard['ifname'] @@ -168,12 +162,12 @@ def apply(wireguard): from vyos.utils.file import write_file text = f'# Automatically generated by interfaces_wireguard.py\nThis file indicates that vyos-domain-resolver service is used by the interfaces_wireguard.\n' - text += "intefaces:\n" + "".join([f" - {peer}\n" for peer in wireguard['peers_need_resolve']]) - Path(domain_resolver_usage).write_text(text) + text += "interfaces:\n" + "".join([f" - {peer}\n" for peer in wireguard['peers_need_resolve']]) write_file(domain_resolver_usage, text) else: - Path(domain_resolver_usage).unlink(missing_ok=True) - if not Path('/run').glob('use-vyos-domain-resolver*'): + if os.path.exists(domain_resolver_usage): + os.unlink(domain_resolver_usage) + if not glob('/run/use-vyos-domain-resolver*'): domain_action = 'stop' call(f'systemctl {domain_action} vyos-domain-resolver.service') diff --git a/src/conf_mode/nat.py b/src/conf_mode/nat.py index 504b3e82a..6c88e5cfd 100755 --- a/src/conf_mode/nat.py +++ b/src/conf_mode/nat.py @@ -16,8 +16,8 @@ import os +from glob import glob from sys import exit -from pathlib import Path from vyos.base import Warning from vyos.config import Config @@ -265,9 +265,9 @@ def apply(nat): text = f'# Automatically generated by nat.py\nThis file indicates that vyos-domain-resolver service is used by nat.\n' write_file(domain_resolver_usage, text) elif os.path.exists(domain_resolver_usage): - Path(domain_resolver_usage).unlink(missing_ok=True) + os.unlink(domain_resolver_usage) - if not Path('/run').glob('use-vyos-domain-resolver*'): + if not glob('/run/use-vyos-domain-resolver*'): domain_action = 'stop' call(f'systemctl {domain_action} vyos-domain-resolver.service') diff --git a/src/conf_mode/policy.py b/src/conf_mode/policy.py index a90e33e81..ec9005890 100755 --- a/src/conf_mode/policy.py +++ b/src/conf_mode/policy.py @@ -14,6 +14,7 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. +import re from sys import exit from vyos.config import Config @@ -24,9 +25,20 @@ from vyos.frrender import get_frrender_dict from vyos.utils.dict import dict_search from vyos.utils.process import is_systemd_service_running from vyos import ConfigError +from vyos.base import Warning from vyos import airbag airbag.enable() +# Sanity checks for large-community-list regex: +# * Require complete 3-tuples, no blank members. Catch missed & doubled colons. +# * Permit appropriate community separators (whitespace, underscore) +# * Permit common regex between tuples while requiring at least one separator +# (eg, "1:1:1_.*_4:4:4", matching "1:1:1 4:4:4" and "1:1:1 2:2:2 4:4:4", +# but not "1:1:13 24:4:4") +# Best practice: stick with basic patterns, mind your wildcards and whitespace. +# Regex that doesn't match this pattern will be allowed with a warning. +large_community_regex_pattern = r'([^: _]+):([^: _]+):([^: _]+)([ _]([^:]+):([^: _]+):([^: _]+))*' + def community_action_compatibility(actions: dict) -> bool: """ Check compatibility of values in community and large community sections @@ -147,6 +159,10 @@ def verify(config_dict): if 'regex' not in rule_config: raise ConfigError(f'A regex {mandatory_error}') + if policy_type == 'large_community_list': + if not re.fullmatch(large_community_regex_pattern, rule_config['regex']): + Warning(f'"policy large-community-list {instance} rule {rule} regex" does not follow expected form and may not match as expected.') + if policy_type in ['prefix_list', 'prefix_list6']: if 'prefix' not in rule_config: raise ConfigError(f'A prefix {mandatory_error}') diff --git a/src/conf_mode/service_monitoring_prometheus.py b/src/conf_mode/service_monitoring_prometheus.py index 9a07d8593..f55b09f6c 100755 --- a/src/conf_mode/service_monitoring_prometheus.py +++ b/src/conf_mode/service_monitoring_prometheus.py @@ -48,9 +48,21 @@ def get_config(config=None): if not conf.exists(base): return None - monitoring = conf.get_config_dict( - base, key_mangling=('-', '_'), get_first_key=True, with_recursive_defaults=True - ) + monitoring = {} + exporters = { + 'node_exporter': base + ['node-exporter'], + 'frr_exporter': base + ['frr-exporter'], + 'blackbox_exporter': base + ['blackbox-exporter'], + } + + for exporter_name, exporter_base in exporters.items(): + if conf.exists(exporter_base): + monitoring[exporter_name] = conf.get_config_dict( + exporter_base, + key_mangling=('-', '_'), + get_first_key=True, + with_recursive_defaults=True, + ) tmp = is_node_changed(conf, base + ['node-exporter', 'vrf']) if tmp: diff --git a/src/conf_mode/service_pppoe-server.py b/src/conf_mode/service_pppoe-server.py index ac697c509..f7cb3dcba 100755 --- a/src/conf_mode/service_pppoe-server.py +++ b/src/conf_mode/service_pppoe-server.py @@ -73,7 +73,9 @@ def get_config(config=None): # https://phabricator.accel-ppp.org/T3 conditions = [is_node_changed(conf, base + ['client-ip-pool']), is_node_changed(conf, base + ['client-ipv6-pool']), - is_node_changed(conf, base + ['interface'])] + is_node_changed(conf, base + ['interface']), + is_node_changed(conf, base + ['authentication','radius','dynamic-author']), + is_node_changed(conf, base + ['authentication','mode'])] if any(conditions): pppoe.update({'restart_required': {}}) pppoe['server_type'] = 'pppoe' diff --git a/src/conf_mode/vpn_ipsec.py b/src/conf_mode/vpn_ipsec.py index 2754314f7..ac25cd671 100755 --- a/src/conf_mode/vpn_ipsec.py +++ b/src/conf_mode/vpn_ipsec.py @@ -727,7 +727,7 @@ def generate(ipsec): for remote_prefix in remote_prefixes: local_net = ipaddress.ip_network(local_prefix) remote_net = ipaddress.ip_network(remote_prefix) - if local_net.overlaps(remote_net): + if local_net.subnet_of(remote_net): if passthrough is None: passthrough = [] passthrough.append(local_prefix) diff --git a/src/conf_mode/vpn_openconnect.py b/src/conf_mode/vpn_openconnect.py index 42785134f..0346c7819 100755 --- a/src/conf_mode/vpn_openconnect.py +++ b/src/conf_mode/vpn_openconnect.py @@ -93,7 +93,7 @@ def verify(ocserv): "radius" in ocserv["authentication"]["mode"]): raise ConfigError('OpenConnect authentication modes are mutually-exclusive, remove either local or radius from your configuration') if "radius" in ocserv["authentication"]["mode"]: - if not ocserv["authentication"]['radius']['server']: + if 'server' not in ocserv['authentication']['radius']: raise ConfigError('OpenConnect authentication mode radius requires at least one RADIUS server') if "local" in ocserv["authentication"]["mode"]: if not ocserv.get("authentication", {}).get("local_users"): diff --git a/src/etc/opennhrp/opennhrp-script.py b/src/etc/opennhrp/opennhrp-script.py deleted file mode 100755 index f6f6d075c..000000000 --- a/src/etc/opennhrp/opennhrp-script.py +++ /dev/null @@ -1,371 +0,0 @@ -#!/usr/bin/env python3 -# -# Copyright (C) 2021-2023 VyOS maintainers and contributors -# -# This program is free software; you can redistribute it and/or modify -# it under the terms of the GNU General Public License version 2 or later as -# published by the Free Software Foundation. -# -# 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, see <http://www.gnu.org/licenses/>. - -import os -import re -import sys -import vyos.ipsec - -from json import loads -from pathlib import Path - -from vyos.logger import getLogger -from vyos.utils.process import cmd -from vyos.utils.process import process_named_running - -NHRP_CONFIG: str = '/run/opennhrp/opennhrp.conf' - - -def vici_get_ipsec_uniqueid(conn: str, src_nbma: str, - dst_nbma: str) -> list[str]: - """ Find and return IKE SAs by src nbma and dst nbma - - Args: - conn (str): a connection name - src_nbma (str): an IP address of NBMA source - dst_nbma (str): an IP address of NBMA destination - - Returns: - list: a list of IKE connections that match a criteria - """ - if not conn or not src_nbma or not dst_nbma: - logger.error( - f'Incomplete input data for resolving IKE unique ids: ' - f'conn: {conn}, src_nbma: {src_nbma}, dst_nbma: {dst_nbma}') - return [] - - try: - logger.info( - f'Resolving IKE unique ids for: conn: {conn}, ' - f'src_nbma: {src_nbma}, dst_nbma: {dst_nbma}') - list_ikeid: list[str] = [] - list_sa: list = vyos.ipsec.get_vici_sas_by_name(conn, None) - for sa in list_sa: - if sa[conn]['local-host'].decode('ascii') == src_nbma \ - and sa[conn]['remote-host'].decode('ascii') == dst_nbma: - list_ikeid.append(sa[conn]['uniqueid'].decode('ascii')) - return list_ikeid - except Exception as err: - logger.error(f'Unable to find unique ids for IKE: {err}') - return [] - - -def vici_ike_terminate(list_ikeid: list[str]) -> bool: - """Terminating IKE SAs by list of IKE IDs - - Args: - list_ikeid (list[str]): a list of IKE ids to terminate - - Returns: - bool: result of termination action - """ - if not list: - logger.warning('An empty list for termination was provided') - return False - - try: - vyos.ipsec.terminate_vici_ikeid_list(list_ikeid) - return True - except Exception as err: - logger.error(f'Failed to terminate SA for IKE ids {list_ikeid}: {err}') - return False - - -def parse_type_ipsec(interface: str) -> tuple[str, str]: - """Get DMVPN Type and NHRP Profile from the configuration - - Args: - interface (str): a name of interface - - Returns: - tuple[str, str]: `peer_type` and `profile_name` - """ - if not interface: - logger.error('Cannot find peer type - no input provided') - return '', '' - - config_file: str = Path(NHRP_CONFIG).read_text() - regex: str = rf'^interface {interface} #(?P<peer_type>hub|spoke) ?(?P<profile_name>[^\n]*)$' - match = re.search(regex, config_file, re.M) - if match: - return match.groupdict()['peer_type'], match.groupdict()[ - 'profile_name'] - return '', '' - - -def add_peer_route(nbma_src: str, nbma_dst: str, mtu: str) -> None: - """Add a route to a NBMA peer - - Args: - nbma_src (str): a local IP address - nbma_dst (str): a remote IP address - mtu (str): a MTU for a route - """ - logger.info(f'Adding route from {nbma_src} to {nbma_dst} with MTU {mtu}') - # Find routes to a peer - route_get_cmd: str = f'sudo ip --json route get {nbma_dst} from {nbma_src}' - try: - route_info_data = loads(cmd(route_get_cmd)) - except Exception as err: - logger.error(f'Unable to find a route to {nbma_dst}: {err}') - return - - # Check if an output has an expected format - if not isinstance(route_info_data, list): - logger.error( - f'Garbage returned from the "{route_get_cmd}" ' - f'command: {route_info_data}') - return - - # Add static routes to a peer - for route_item in route_info_data: - route_dev = route_item.get('dev') - route_dst = route_item.get('dst') - route_gateway = route_item.get('gateway') - # Prepare a command to add a route - route_add_cmd = 'sudo ip route add' - if route_dst: - route_add_cmd = f'{route_add_cmd} {route_dst}' - if route_gateway: - route_add_cmd = f'{route_add_cmd} via {route_gateway}' - if route_dev: - route_add_cmd = f'{route_add_cmd} dev {route_dev}' - route_add_cmd = f'{route_add_cmd} proto 42 mtu {mtu}' - # Add a route - try: - cmd(route_add_cmd) - except Exception as err: - logger.error( - f'Unable to add a route using command "{route_add_cmd}": ' - f'{err}') - - -def vici_initiate(conn: str, child_sa: str, src_addr: str, - dest_addr: str) -> bool: - """Initiate IKE SA connection with specific peer - - Args: - conn (str): an IKE connection name - child_sa (str): a child SA profile name - src_addr (str): NBMA local address - dest_addr (str): NBMA address of a peer - - Returns: - bool: a result of initiation command - """ - logger.info( - f'Trying to initiate connection. Name: {conn}, child sa: {child_sa}, ' - f'src_addr: {src_addr}, dst_addr: {dest_addr}') - try: - vyos.ipsec.vici_initiate(conn, child_sa, src_addr, dest_addr) - return True - except Exception as err: - logger.error(f'Unable to initiate connection {err}') - return False - - -def vici_terminate(conn: str, src_addr: str, dest_addr: str) -> None: - """Find and terminate IKE SAs by local NBMA and remote NBMA addresses - - Args: - conn (str): IKE connection name - src_addr (str): NBMA local address - dest_addr (str): NBMA address of a peer - """ - logger.info( - f'Terminating IKE connection {conn} between {src_addr} ' - f'and {dest_addr}') - - ikeid_list: list[str] = vici_get_ipsec_uniqueid(conn, src_addr, dest_addr) - - if not ikeid_list: - logger.warning( - f'No active sessions found for IKE profile {conn}, ' - f'local NBMA {src_addr}, remote NBMA {dest_addr}') - else: - try: - vyos.ipsec.terminate_vici_ikeid_list(ikeid_list) - except Exception as err: - logger.error( - f'Failed to terminate SA for IKE ids {ikeid_list}: {err}') - -def iface_up(interface: str) -> None: - """Proceed tunnel interface UP event - - Args: - interface (str): an interface name - """ - if not interface: - logger.warning('No interface name provided for UP event') - - logger.info(f'Turning up interface {interface}') - try: - cmd(f'sudo ip route flush proto 42 dev {interface}') - cmd(f'sudo ip neigh flush dev {interface}') - except Exception as err: - logger.error( - f'Unable to flush route on interface "{interface}": {err}') - - -def peer_up(dmvpn_type: str, conn: str) -> None: - """Proceed NHRP peer UP event - - Args: - dmvpn_type (str): a type of peer - conn (str): an IKE profile name - """ - logger.info(f'Peer UP event for {dmvpn_type} using IKE profile {conn}') - src_nbma = os.getenv('NHRP_SRCNBMA') - dest_nbma = os.getenv('NHRP_DESTNBMA') - dest_mtu = os.getenv('NHRP_DESTMTU') - - if not src_nbma or not dest_nbma: - logger.error( - f'Can not get NHRP NBMA addresses: local {src_nbma}, ' - f'remote {dest_nbma}') - return - - logger.info(f'NBMA addresses: local {src_nbma}, remote {dest_nbma}') - if dest_mtu: - add_peer_route(src_nbma, dest_nbma, dest_mtu) - if conn and dmvpn_type == 'spoke' and process_named_running('charon'): - vici_terminate(conn, src_nbma, dest_nbma) - vici_initiate(conn, 'dmvpn', src_nbma, dest_nbma) - - -def peer_down(dmvpn_type: str, conn: str) -> None: - """Proceed NHRP peer DOWN event - - Args: - dmvpn_type (str): a type of peer - conn (str): an IKE profile name - """ - logger.info(f'Peer DOWN event for {dmvpn_type} using IKE profile {conn}') - - src_nbma = os.getenv('NHRP_SRCNBMA') - dest_nbma = os.getenv('NHRP_DESTNBMA') - - if not src_nbma or not dest_nbma: - logger.error( - f'Can not get NHRP NBMA addresses: local {src_nbma}, ' - f'remote {dest_nbma}') - return - - logger.info(f'NBMA addresses: local {src_nbma}, remote {dest_nbma}') - if conn and dmvpn_type == 'spoke' and process_named_running('charon'): - vici_terminate(conn, src_nbma, dest_nbma) - try: - cmd(f'sudo ip route del {dest_nbma} src {src_nbma} proto 42') - except Exception as err: - logger.error( - f'Unable to del route from {src_nbma} to {dest_nbma}: {err}') - - -def route_up(interface: str) -> None: - """Proceed NHRP route UP event - - Args: - interface (str): an interface name - """ - logger.info(f'Route UP event for interface {interface}') - - dest_addr = os.getenv('NHRP_DESTADDR') - dest_prefix = os.getenv('NHRP_DESTPREFIX') - next_hop = os.getenv('NHRP_NEXTHOP') - - if not dest_addr or not dest_prefix or not next_hop: - logger.error( - f'Can not get route details: dest_addr {dest_addr}, ' - f'dest_prefix {dest_prefix}, next_hop {next_hop}') - return - - logger.info( - f'Route details: dest_addr {dest_addr}, dest_prefix {dest_prefix}, ' - f'next_hop {next_hop}') - - try: - cmd(f'sudo ip route replace {dest_addr}/{dest_prefix} proto 42 \ - via {next_hop} dev {interface}') - cmd('sudo ip route flush cache') - except Exception as err: - logger.error( - f'Unable replace or flush route to {dest_addr}/{dest_prefix} ' - f'via {next_hop} dev {interface}: {err}') - - -def route_down(interface: str) -> None: - """Proceed NHRP route DOWN event - - Args: - interface (str): an interface name - """ - logger.info(f'Route DOWN event for interface {interface}') - - dest_addr = os.getenv('NHRP_DESTADDR') - dest_prefix = os.getenv('NHRP_DESTPREFIX') - - if not dest_addr or not dest_prefix: - logger.error( - f'Can not get route details: dest_addr {dest_addr}, ' - f'dest_prefix {dest_prefix}') - return - - logger.info( - f'Route details: dest_addr {dest_addr}, dest_prefix {dest_prefix}') - try: - cmd(f'sudo ip route del {dest_addr}/{dest_prefix} proto 42') - cmd('sudo ip route flush cache') - except Exception as err: - logger.error( - f'Unable delete or flush route to {dest_addr}/{dest_prefix}: ' - f'{err}') - - -if __name__ == '__main__': - logger = getLogger('opennhrp-script', syslog=True) - logger.debug( - f'Running script with arguments: {sys.argv}, ' - f'environment: {os.environ}') - - action = sys.argv[1] - interface = os.getenv('NHRP_INTERFACE') - - if not interface: - logger.error('Can not get NHRP interface name') - sys.exit(1) - - dmvpn_type, profile_name = parse_type_ipsec(interface) - if not dmvpn_type: - logger.info(f'Interface {interface} is not NHRP tunnel') - sys.exit() - - dmvpn_conn: str = '' - if profile_name: - dmvpn_conn: str = f'dmvpn-{profile_name}-{interface}' - if action == 'interface-up': - iface_up(interface) - elif action == 'peer-register': - pass - elif action == 'peer-up': - peer_up(dmvpn_type, dmvpn_conn) - elif action == 'peer-down': - peer_down(dmvpn_type, dmvpn_conn) - elif action == 'route-up': - route_up(interface) - elif action == 'route-down': - route_down(interface) - - sys.exit() diff --git a/src/systemd/opennhrp.service b/src/systemd/opennhrp.service deleted file mode 100644 index c9a44de29..000000000 --- a/src/systemd/opennhrp.service +++ /dev/null @@ -1,13 +0,0 @@ -[Unit] -Description=OpenNHRP -After=vyos-router.service -ConditionPathExists=/run/opennhrp/opennhrp.conf -StartLimitIntervalSec=0 - -[Service] -Type=forking -ExecStart=/usr/sbin/opennhrp -d -v -a /run/opennhrp.socket -c /run/opennhrp/opennhrp.conf -s /etc/opennhrp/opennhrp-script.py -p /run/opennhrp/opennhrp.pid -ExecReload=/usr/bin/kill -HUP $MAINPID -PIDFile=/run/opennhrp/opennhrp.pid -Restart=on-failure -RestartSec=20 diff --git a/src/validators/bgp-large-community-list b/src/validators/bgp-large-community-list index 9ba5b27eb..75276630c 100755 --- a/src/validators/bgp-large-community-list +++ b/src/validators/bgp-large-community-list @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -# Copyright (C) 2021-2023 VyOS maintainers and contributors +# Copyright (C) 2021-2025 VyOS maintainers and contributors # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 or later as @@ -17,18 +17,27 @@ import re import sys -pattern = '(.*):(.*):(.*)' -allowedChars = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '.', '+', '*', '?', '^', '$', '(', ')', '[', ']', '{', '}', '|', '\\', ':', '-' } +allowedChars = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '.', '+', '*', '?', '^', '$', '(', ')', '[', ']', '{', '}', '|', '\\', ':', '-', '_', ' ' } if __name__ == '__main__': if len(sys.argv) != 2: sys.exit(1) - value = sys.argv[1].split(':') - if not len(value) == 3: + value = sys.argv[1] + + # Require at least one well-formed large-community tuple in the pattern. + tmp = value.split(':') + if len(tmp) < 3: + sys.exit(1) + + # Simple guard against invalid community & 1003.2 pattern chars + if not set(value).issubset(allowedChars): sys.exit(1) - if not (re.match(pattern, sys.argv[1]) and set(sys.argv[1]).issubset(allowedChars)): + # Don't feed FRR badly formed regex + try: + re.compile(value) + except re.error: sys.exit(1) sys.exit(0) |