diff --git a/interface-definitions/include/firewall/common-rule-inet.xml.i b/interface-definitions/include/firewall/common-rule-inet.xml.i
index 872abe6cc..a55a1a551 100644
--- a/interface-definitions/include/firewall/common-rule-inet.xml.i
+++ b/interface-definitions/include/firewall/common-rule-inet.xml.i
@@ -1,377 +1,378 @@
<!-- include start from firewall/common-rule-inet.xml.i -->
#include <include/firewall/action.xml.i>
#include <include/generic-description.xml.i>
#include <include/firewall/dscp.xml.i>
#include <include/firewall/packet-options.xml.i>
+#include <include/firewall/firewall-mark.xml.i>
#include <include/firewall/connection-mark.xml.i>
#include <include/firewall/conntrack-helper.xml.i>
#include <include/firewall/nft-queue.xml.i>
<leafNode name="disable">
<properties>
<help>Option to disable firewall rule</help>
<valueless/>
</properties>
</leafNode>
<node name="fragment">
<properties>
<help>IP fragment match</help>
</properties>
<children>
<leafNode name="match-frag">
<properties>
<help>Second and further fragments of fragmented packets</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="match-non-frag">
<properties>
<help>Head fragments or unfragmented packets</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<node name="ipsec">
<properties>
<help>Inbound IPsec packets</help>
</properties>
<children>
<leafNode name="match-ipsec">
<properties>
<help>Inbound IPsec packets</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="match-none">
<properties>
<help>Inbound non-IPsec packets</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<node name="limit">
<properties>
<help>Rate limit using a token bucket filter</help>
</properties>
<children>
<leafNode name="burst">
<properties>
<help>Maximum number of packets to allow in excess of rate</help>
<valueHelp>
<format>u32:0-4294967295</format>
<description>Maximum number of packets to allow in excess of rate</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-4294967295"/>
</constraint>
</properties>
</leafNode>
<leafNode name="rate">
<properties>
<help>Maximum average matching rate</help>
<valueHelp>
<format>txt</format>
<description>integer/unit (Example: 5/minute)</description>
</valueHelp>
<constraint>
<regex>\d+/(second|minute|hour|day)</regex>
</constraint>
</properties>
</leafNode>
</children>
</node>
<leafNode name="log">
<properties>
<help>Option to log packets matching rule</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable log</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable log</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="log">
<properties>
<help>Option to log packets matching rule</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable log</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable log</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
#include <include/firewall/rule-log-options.xml.i>
<node name="connection-status">
<properties>
<help>Connection status</help>
</properties>
<children>
<leafNode name="nat">
<properties>
<help>NAT connection status</help>
<completionHelp>
<list>destination source</list>
</completionHelp>
<valueHelp>
<format>destination</format>
<description>Match connections that are subject to destination NAT</description>
</valueHelp>
<valueHelp>
<format>source</format>
<description>Match connections that are subject to source NAT</description>
</valueHelp>
<constraint>
<regex>(destination|source)</regex>
</constraint>
</properties>
</leafNode>
</children>
</node>
<leafNode name="protocol">
<properties>
<help>Protocol to match (protocol name, number, or "all")</help>
<completionHelp>
<script>${vyos_completion_dir}/list_protocols.sh</script>
<list>all tcp_udp</list>
</completionHelp>
<valueHelp>
<format>all</format>
<description>All IP protocols</description>
</valueHelp>
<valueHelp>
<format>tcp_udp</format>
<description>Both TCP and UDP</description>
</valueHelp>
<valueHelp>
<format>u32:0-255</format>
<description>IP protocol number</description>
</valueHelp>
<valueHelp>
<format><protocol></format>
<description>IP protocol name</description>
</valueHelp>
<valueHelp>
<format>!<protocol></format>
<description>IP protocol name</description>
</valueHelp>
<constraint>
<validator name="ip-protocol"/>
</constraint>
</properties>
</leafNode>
<node name="recent">
<properties>
<help>Parameters for matching recently seen sources</help>
</properties>
<children>
<leafNode name="count">
<properties>
<help>Source addresses seen more than N times</help>
<valueHelp>
<format>u32:1-255</format>
<description>Source addresses seen more than N times</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-255"/>
</constraint>
</properties>
</leafNode>
<leafNode name="time">
<properties>
<help>Source addresses seen in the last second/minute/hour</help>
<completionHelp>
<list>second minute hour</list>
</completionHelp>
<valueHelp>
<format>second</format>
<description>Source addresses seen COUNT times in the last second</description>
</valueHelp>
<valueHelp>
<format>minute</format>
<description>Source addresses seen COUNT times in the last minute</description>
</valueHelp>
<valueHelp>
<format>hour</format>
<description>Source addresses seen COUNT times in the last hour</description>
</valueHelp>
<constraint>
<regex>(second|minute|hour)</regex>
</constraint>
</properties>
</leafNode>
</children>
</node>
#include <include/firewall/synproxy.xml.i>
<node name="state">
<properties>
<help>Session state</help>
</properties>
<children>
<leafNode name="established">
<properties>
<help>Established state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="invalid">
<properties>
<help>Invalid state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="new">
<properties>
<help>New state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="related">
<properties>
<help>Related state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
</children>
</node>
#include <include/firewall/tcp-flags.xml.i>
#include <include/firewall/tcp-mss.xml.i>
<node name="time">
<properties>
<help>Time to match rule</help>
</properties>
<children>
<leafNode name="startdate">
<properties>
<help>Date to start matching rule</help>
<valueHelp>
<format>txt</format>
<description>Enter date using following notation - YYYY-MM-DD</description>
</valueHelp>
<constraint>
<regex>(\d{4}\-\d{2}\-\d{2})</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="starttime">
<properties>
<help>Time of day to start matching rule</help>
<valueHelp>
<format>txt</format>
<description>Enter time using using 24 hour notation - hh:mm:ss</description>
</valueHelp>
<constraint>
<regex>([0-2][0-9](\:[0-5][0-9]){1,2})</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="stopdate">
<properties>
<help>Date to stop matching rule</help>
<valueHelp>
<format>txt</format>
<description>Enter date using following notation - YYYY-MM-DD</description>
</valueHelp>
<constraint>
<regex>(\d{4}\-\d{2}\-\d{2})</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="stoptime">
<properties>
<help>Time of day to stop matching rule</help>
<valueHelp>
<format>txt</format>
<description>Enter time using using 24 hour notation - hh:mm:ss</description>
</valueHelp>
<constraint>
<regex>([0-2][0-9](\:[0-5][0-9]){1,2})</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="weekdays">
<properties>
<help>Comma separated weekdays to match rule on</help>
<valueHelp>
<format>txt</format>
<description>Name of day (Monday, Tuesday, Wednesday, Thursdays, Friday, Saturday, Sunday)</description>
</valueHelp>
<valueHelp>
<format>u32:0-6</format>
<description>Day number (0 = Sunday ... 6 = Saturday)</description>
</valueHelp>
</properties>
</leafNode>
</children>
</node>
<!-- include end -->
diff --git a/interface-definitions/include/firewall/firewall-mark.xml.i b/interface-definitions/include/firewall/firewall-mark.xml.i
new file mode 100644
index 000000000..36a939ba3
--- /dev/null
+++ b/interface-definitions/include/firewall/firewall-mark.xml.i
@@ -0,0 +1,26 @@
+<!-- include start from firewall/firewall-mark.xml.i -->
+<leafNode name="mark">
+ <properties>
+ <help>Firewall mark</help>
+ <valueHelp>
+ <format>u32:0-2147483647</format>
+ <description>Firewall mark to match</description>
+ </valueHelp>
+ <valueHelp>
+ <format>!u32:0-2147483647</format>
+ <description>Inverted Firewall mark to match</description>
+ </valueHelp>
+ <valueHelp>
+ <format><start-end></format>
+ <description>Firewall mark range to match</description>
+ </valueHelp>
+ <valueHelp>
+ <format>!<start-end></format>
+ <description>Firewall mark inverted range to match</description>
+ </valueHelp>
+ <constraint>
+ <validator name="numeric-exclude" argument="--allow-range --range 0-2147483647"/>
+ </constraint>
+ </properties>
+</leafNode>
+<!-- include end -->
\ No newline at end of file
diff --git a/interface-definitions/include/policy/route-common.xml.i b/interface-definitions/include/policy/route-common.xml.i
index 6551d23ab..8eab04d4a 100644
--- a/interface-definitions/include/policy/route-common.xml.i
+++ b/interface-definitions/include/policy/route-common.xml.i
@@ -1,361 +1,362 @@
<!-- include start from policy/route-common.xml.i -->
#include <include/policy/route-rule-action.xml.i>
#include <include/generic-description.xml.i>
+#include <include/firewall/firewall-mark.xml.i>
<leafNode name="disable">
<properties>
<help>Option to disable firewall rule</help>
<valueless/>
</properties>
</leafNode>
<node name="fragment">
<properties>
<help>IP fragment match</help>
</properties>
<children>
<leafNode name="match-frag">
<properties>
<help>Second and further fragments of fragmented packets</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="match-non-frag">
<properties>
<help>Head fragments or unfragmented packets</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<node name="ipsec">
<properties>
<help>Inbound IPsec packets</help>
</properties>
<children>
<leafNode name="match-ipsec">
<properties>
<help>Inbound IPsec packets</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="match-none">
<properties>
<help>Inbound non-IPsec packets</help>
<valueless/>
</properties>
</leafNode>
</children>
</node>
<node name="limit">
<properties>
<help>Rate limit using a token bucket filter</help>
</properties>
<children>
<leafNode name="burst">
<properties>
<help>Maximum number of packets to allow in excess of rate</help>
<valueHelp>
<format>u32:0-4294967295</format>
<description>Maximum number of packets to allow in excess of rate</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-4294967295"/>
</constraint>
</properties>
</leafNode>
<leafNode name="rate">
<properties>
<help>Maximum average matching rate</help>
<valueHelp>
<format>u32:0-4294967295</format>
<description>Maximum average matching rate</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-4294967295"/>
</constraint>
</properties>
</leafNode>
</children>
</node>
<leafNode name="log">
<properties>
<help>Option to log packets matching rule</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable log</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable log</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="protocol">
<properties>
<help>Protocol to match (protocol name, number, or "all")</help>
<completionHelp>
<script>cat /etc/protocols | sed -e '/^#.*/d' | awk '{ print $1 }'</script>
</completionHelp>
<valueHelp>
<format>all</format>
<description>All IP protocols</description>
</valueHelp>
<valueHelp>
<format>tcp_udp</format>
<description>Both TCP and UDP</description>
</valueHelp>
<valueHelp>
<format>0-255</format>
<description>IP protocol number</description>
</valueHelp>
<valueHelp>
<format>!<protocol></format>
<description>IP protocol number</description>
</valueHelp>
<constraint>
<validator name="ip-protocol"/>
</constraint>
</properties>
<defaultValue>all</defaultValue>
</leafNode>
<node name="recent">
<properties>
<help>Parameters for matching recently seen sources</help>
</properties>
<children>
<leafNode name="count">
<properties>
<help>Source addresses seen more than N times</help>
<valueHelp>
<format>u32:1-255</format>
<description>Source addresses seen more than N times</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-255"/>
</constraint>
</properties>
</leafNode>
<leafNode name="time">
<properties>
<help>Source addresses seen in the last N seconds</help>
<valueHelp>
<format>u32:0-4294967295</format>
<description>Source addresses seen in the last N seconds</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-4294967295"/>
</constraint>
</properties>
</leafNode>
</children>
</node>
<node name="set">
<properties>
<help>Packet modifications</help>
</properties>
<children>
<leafNode name="connection-mark">
<properties>
<help>Connection marking</help>
<valueHelp>
<format>u32:0-2147483647</format>
<description>Connection marking</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-2147483647"/>
</constraint>
</properties>
</leafNode>
<leafNode name="dscp">
<properties>
<help>Packet Differentiated Services Codepoint (DSCP)</help>
<valueHelp>
<format>u32:0-63</format>
<description>DSCP number</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 0-63"/>
</constraint>
</properties>
</leafNode>
<leafNode name="mark">
<properties>
<help>Packet marking</help>
<valueHelp>
<format>u32:1-2147483647</format>
<description>Packet marking</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-2147483647"/>
</constraint>
</properties>
</leafNode>
<leafNode name="table">
<properties>
<help>Routing table to forward packet with</help>
<valueHelp>
<format>u32:1-200</format>
<description>Table number</description>
</valueHelp>
<valueHelp>
<format>main</format>
<description>Main table</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 1-200"/>
<regex>(main)</regex>
</constraint>
<completionHelp>
<list>main</list>
<path>protocols static table</path>
</completionHelp>
</properties>
</leafNode>
<leafNode name="tcp-mss">
<properties>
<help>TCP Maximum Segment Size</help>
<valueHelp>
<format>u32:500-1460</format>
<description>Explicitly set TCP MSS value</description>
</valueHelp>
<constraint>
<validator name="numeric" argument="--range 500-1460"/>
</constraint>
</properties>
</leafNode>
</children>
</node>
<node name="state">
<properties>
<help>Session state</help>
</properties>
<children>
<leafNode name="established">
<properties>
<help>Established state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="invalid">
<properties>
<help>Invalid state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="new">
<properties>
<help>New state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
<leafNode name="related">
<properties>
<help>Related state</help>
<completionHelp>
<list>enable disable</list>
</completionHelp>
<valueHelp>
<format>enable</format>
<description>Enable</description>
</valueHelp>
<valueHelp>
<format>disable</format>
<description>Disable</description>
</valueHelp>
<constraint>
<regex>(enable|disable)</regex>
</constraint>
</properties>
</leafNode>
</children>
</node>
#include <include/firewall/tcp-flags.xml.i>
#include <include/firewall/tcp-mss.xml.i>
<node name="time">
<properties>
<help>Time to match rule</help>
</properties>
<children>
<leafNode name="monthdays">
<properties>
<help>Monthdays to match rule on</help>
</properties>
</leafNode>
<leafNode name="startdate">
<properties>
<help>Date to start matching rule</help>
</properties>
</leafNode>
<leafNode name="starttime">
<properties>
<help>Time of day to start matching rule</help>
</properties>
</leafNode>
<leafNode name="stopdate">
<properties>
<help>Date to stop matching rule</help>
</properties>
</leafNode>
<leafNode name="stoptime">
<properties>
<help>Time of day to stop matching rule</help>
</properties>
</leafNode>
<leafNode name="utc">
<properties>
<help>Interpret times for startdate, stopdate, starttime and stoptime to be UTC</help>
<valueless/>
</properties>
</leafNode>
<leafNode name="weekdays">
<properties>
<help>Weekdays to match rule on</help>
</properties>
</leafNode>
</children>
</node>
<!-- include end -->
diff --git a/python/vyos/firewall.py b/python/vyos/firewall.py
index 9122e264e..c07ed1adf 100644
--- a/python/vyos/firewall.py
+++ b/python/vyos/firewall.py
@@ -1,618 +1,626 @@
#!/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 csv
import gzip
import os
import re
from pathlib import Path
from socket import AF_INET
from socket import AF_INET6
from socket import getaddrinfo
from time import strftime
from vyos.remote import download
from vyos.template import is_ipv4
from vyos.template import render
from vyos.utils.dict import dict_search_args
from vyos.utils.dict import dict_search_recursive
from vyos.utils.process import call
from vyos.utils.process import cmd
from vyos.utils.process import run
# Domain Resolver
def fqdn_config_parse(firewall):
firewall['ip_fqdn'] = {}
firewall['ip6_fqdn'] = {}
for domain, path in dict_search_recursive(firewall, 'fqdn'):
hook_name = path[1]
priority = path[2]
fw_name = path[2]
rule = path[4]
suffix = path[5][0]
set_name = f'{hook_name}_{priority}_{rule}_{suffix}'
if (path[0] == 'ipv4') and (path[1] == 'forward' or path[1] == 'input' or path[1] == 'output' or path[1] == 'name'):
firewall['ip_fqdn'][set_name] = domain
elif (path[0] == 'ipv6') and (path[1] == 'forward' or path[1] == 'input' or path[1] == 'output' or path[1] == 'name'):
if path[1] == 'name':
set_name = f'name6_{priority}_{rule}_{suffix}'
firewall['ip6_fqdn'][set_name] = domain
def fqdn_resolve(fqdn, ipv6=False):
try:
res = getaddrinfo(fqdn, None, AF_INET6 if ipv6 else AF_INET)
return set(item[4][0] for item in res)
except:
return None
# End Domain Resolver
def find_nftables_rule(table, chain, rule_matches=[]):
# Find rule in table/chain that matches all criteria and return the handle
results = cmd(f'sudo nft -a list chain {table} {chain}').split("\n")
for line in results:
if all(rule_match in line for rule_match in rule_matches):
handle_search = re.search('handle (\d+)', line)
if handle_search:
return handle_search[1]
return None
def remove_nftables_rule(table, chain, handle):
cmd(f'sudo nft delete rule {table} {chain} handle {handle}')
# Functions below used by template generation
def nft_action(vyos_action):
if vyos_action == 'accept':
return 'return'
return vyos_action
def parse_rule(rule_conf, hook, fw_name, rule_id, ip_name):
output = []
#def_suffix = '6' if ip_name == 'ip6' else ''
if ip_name == 'ip6':
def_suffix = '6'
family = 'ipv6'
else:
def_suffix = ''
family = 'bri' if ip_name == 'bri' else 'ipv4'
if 'state' in rule_conf and rule_conf['state']:
states = ",".join([s for s, v in rule_conf['state'].items() if v == 'enable'])
if states:
output.append(f'ct state {{{states}}}')
if 'conntrack_helper' in rule_conf:
helper_map = {'h323': ['RAS', 'Q.931'], 'nfs': ['rpc'], 'sqlnet': ['tns']}
helper_out = []
for helper in rule_conf['conntrack_helper']:
if helper in helper_map:
helper_out.extend(helper_map[helper])
else:
helper_out.append(helper)
if helper_out:
helper_str = ','.join(f'"{s}"' for s in helper_out)
output.append(f'ct helper {{{helper_str}}}')
if 'connection_status' in rule_conf and rule_conf['connection_status']:
status = rule_conf['connection_status']
if status['nat'] == 'destination':
nat_status = '{dnat}'
output.append(f'ct status {nat_status}')
if status['nat'] == 'source':
nat_status = '{snat}'
output.append(f'ct status {nat_status}')
if 'protocol' in rule_conf and rule_conf['protocol'] != 'all':
proto = rule_conf['protocol']
operator = ''
if proto[0] == '!':
operator = '!='
proto = proto[1:]
if proto == 'tcp_udp':
proto = '{tcp, udp}'
output.append(f'meta l4proto {operator} {proto}')
for side in ['destination', 'source']:
if side in rule_conf:
prefix = side[0]
side_conf = rule_conf[side]
address_mask = side_conf.get('address_mask', None)
if 'address' in side_conf:
suffix = side_conf['address']
operator = ''
exclude = suffix[0] == '!'
if exclude:
operator = '!= '
suffix = suffix[1:]
if address_mask:
operator = '!=' if exclude else '=='
operator = f'& {address_mask} {operator} '
output.append(f'{ip_name} {prefix}addr {operator}{suffix}')
if 'fqdn' in side_conf:
fqdn = side_conf['fqdn']
hook_name = ''
operator = ''
if fqdn[0] == '!':
operator = '!='
if hook == 'FWD':
hook_name = 'forward'
if hook == 'INP':
hook_name = 'input'
if hook == 'OUT':
hook_name = 'output'
if hook == 'NAM':
hook_name = f'name{def_suffix}'
output.append(f'{ip_name} {prefix}addr {operator} @FQDN_{hook_name}_{fw_name}_{rule_id}_{prefix}')
if dict_search_args(side_conf, 'geoip', 'country_code'):
operator = ''
hook_name = ''
if dict_search_args(side_conf, 'geoip', 'inverse_match') != None:
operator = '!='
if hook == 'FWD':
hook_name = 'forward'
if hook == 'INP':
hook_name = 'input'
if hook == 'OUT':
hook_name = 'output'
if hook == 'NAM':
hook_name = f'name'
output.append(f'{ip_name} {prefix}addr {operator} @GEOIP_CC{def_suffix}_{hook_name}_{fw_name}_{rule_id}')
if 'mac_address' in side_conf:
suffix = side_conf["mac_address"]
if suffix[0] == '!':
suffix = f'!= {suffix[1:]}'
output.append(f'ether {prefix}addr {suffix}')
if 'port' in side_conf:
proto = rule_conf['protocol']
port = side_conf['port'].split(',')
ports = []
negated_ports = []
for p in port:
if p[0] == '!':
negated_ports.append(p[1:])
else:
ports.append(p)
if proto == 'tcp_udp':
proto = 'th'
if ports:
ports_str = ','.join(ports)
output.append(f'{proto} {prefix}port {{{ports_str}}}')
if negated_ports:
negated_ports_str = ','.join(negated_ports)
output.append(f'{proto} {prefix}port != {{{negated_ports_str}}}')
if 'group' in side_conf:
group = side_conf['group']
if 'address_group' in group:
group_name = group['address_group']
operator = ''
exclude = group_name[0] == "!"
if exclude:
operator = '!='
group_name = group_name[1:]
if address_mask:
operator = '!=' if exclude else '=='
operator = f'& {address_mask} {operator}'
output.append(f'{ip_name} {prefix}addr {operator} @A{def_suffix}_{group_name}')
# Generate firewall group domain-group
elif 'domain_group' in group:
group_name = group['domain_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_name} {prefix}addr {operator} @D_{group_name}')
elif 'network_group' in group:
group_name = group['network_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{ip_name} {prefix}addr {operator} @N{def_suffix}_{group_name}')
if 'mac_group' in group:
group_name = group['mac_group']
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'ether {prefix}addr {operator} @M_{group_name}')
if 'port_group' in group:
proto = rule_conf['protocol']
group_name = group['port_group']
if proto == 'tcp_udp':
proto = 'th'
operator = ''
if group_name[0] == '!':
operator = '!='
group_name = group_name[1:]
output.append(f'{proto} {prefix}port {operator} @P_{group_name}')
if dict_search_args(rule_conf, 'action') == 'synproxy':
output.append('ct state invalid,untracked')
if 'hop_limit' in rule_conf:
operators = {'eq': '==', 'gt': '>', 'lt': '<'}
for op, operator in operators.items():
if op in rule_conf['hop_limit']:
value = rule_conf['hop_limit'][op]
output.append(f'ip6 hoplimit {operator} {value}')
if 'inbound_interface' in rule_conf:
operator = ''
if 'interface_name' in rule_conf['inbound_interface']:
iiface = rule_conf['inbound_interface']['interface_name']
if iiface[0] == '!':
operator = '!='
iiface = iiface[1:]
output.append(f'iifname {operator} {{{iiface}}}')
else:
iiface = rule_conf['inbound_interface']['interface_group']
if iiface[0] == '!':
operator = '!='
iiface = iiface[1:]
output.append(f'iifname {operator} @I_{iiface}')
if 'outbound_interface' in rule_conf:
operator = ''
if 'interface_name' in rule_conf['outbound_interface']:
oiface = rule_conf['outbound_interface']['interface_name']
if oiface[0] == '!':
operator = '!='
oiface = oiface[1:]
output.append(f'oifname {operator} {{{oiface}}}')
else:
oiface = rule_conf['outbound_interface']['interface_group']
if oiface[0] == '!':
operator = '!='
oiface = oiface[1:]
output.append(f'oifname {operator} @I_{oiface}')
if 'ttl' in rule_conf:
operators = {'eq': '==', 'gt': '>', 'lt': '<'}
for op, operator in operators.items():
if op in rule_conf['ttl']:
value = rule_conf['ttl'][op]
output.append(f'ip ttl {operator} {value}')
for icmp in ['icmp', 'icmpv6']:
if icmp in rule_conf:
if 'type_name' in rule_conf[icmp]:
output.append(icmp + ' type ' + rule_conf[icmp]['type_name'])
else:
if 'code' in rule_conf[icmp]:
output.append(icmp + ' code ' + rule_conf[icmp]['code'])
if 'type' in rule_conf[icmp]:
output.append(icmp + ' type ' + rule_conf[icmp]['type'])
if 'packet_length' in rule_conf:
lengths_str = ','.join(rule_conf['packet_length'])
output.append(f'ip{def_suffix} length {{{lengths_str}}}')
if 'packet_length_exclude' in rule_conf:
negated_lengths_str = ','.join(rule_conf['packet_length_exclude'])
output.append(f'ip{def_suffix} length != {{{negated_lengths_str}}}')
if 'packet_type' in rule_conf:
output.append(f'pkttype ' + rule_conf['packet_type'])
if 'dscp' in rule_conf:
dscp_str = ','.join(rule_conf['dscp'])
output.append(f'ip{def_suffix} dscp {{{dscp_str}}}')
if 'dscp_exclude' in rule_conf:
negated_dscp_str = ','.join(rule_conf['dscp_exclude'])
output.append(f'ip{def_suffix} dscp != {{{negated_dscp_str}}}')
if 'ipsec' in rule_conf:
if 'match_ipsec' in rule_conf['ipsec']:
output.append('meta ipsec == 1')
if 'match_none' in rule_conf['ipsec']:
output.append('meta ipsec == 0')
if 'fragment' in rule_conf:
# Checking for fragmentation after priority -400 is not possible,
# so we use a priority -450 hook to set a mark
if 'match_frag' in rule_conf['fragment']:
output.append('meta mark 0xffff1')
if 'match_non_frag' in rule_conf['fragment']:
output.append('meta mark != 0xffff1')
if 'limit' in rule_conf:
if 'rate' in rule_conf['limit']:
output.append(f'limit rate {rule_conf["limit"]["rate"]}')
if 'burst' in rule_conf['limit']:
output.append(f'burst {rule_conf["limit"]["burst"]} packets')
if 'recent' in rule_conf:
count = rule_conf['recent']['count']
time = rule_conf['recent']['time']
output.append(f'add @RECENT{def_suffix}_{hook}_{fw_name}_{rule_id} {{ {ip_name} saddr limit rate over {count}/{time} burst {count} packets }}')
if 'time' in rule_conf:
output.append(parse_time(rule_conf['time']))
tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags')
if tcp_flags:
output.append(parse_tcp_flags(tcp_flags))
# TCP MSS
tcp_mss = dict_search_args(rule_conf, 'tcp', 'mss')
if tcp_mss:
output.append(f'tcp option maxseg size {tcp_mss}')
if 'connection_mark' in rule_conf:
conn_mark_str = ','.join(rule_conf['connection_mark'])
output.append(f'ct mark {{{conn_mark_str}}}')
+ if 'mark' in rule_conf:
+ mark = rule_conf['mark']
+ operator = ''
+ if mark[0] == '!':
+ operator = '!='
+ mark = mark[1:]
+ output.append(f'meta mark {operator} {{{mark}}}')
+
if 'vlan' in rule_conf:
if 'id' in rule_conf['vlan']:
output.append(f'vlan id {rule_conf["vlan"]["id"]}')
if 'priority' in rule_conf['vlan']:
output.append(f'vlan pcp {rule_conf["vlan"]["priority"]}')
if 'log' in rule_conf and rule_conf['log'] == 'enable':
action = rule_conf['action'] if 'action' in rule_conf else 'accept'
#output.append(f'log prefix "[{fw_name[:19]}-{rule_id}-{action[:1].upper()}]"')
output.append(f'log prefix "[{family}-{hook}-{fw_name}-{rule_id}-{action[:1].upper()}]"')
##{family}-{hook}-{fw_name}-{rule_id}
if 'log_options' in rule_conf:
if 'level' in rule_conf['log_options']:
log_level = rule_conf['log_options']['level']
output.append(f'log level {log_level}')
if 'group' in rule_conf['log_options']:
log_group = rule_conf['log_options']['group']
output.append(f'log group {log_group}')
if 'queue_threshold' in rule_conf['log_options']:
queue_threshold = rule_conf['log_options']['queue_threshold']
output.append(f'queue-threshold {queue_threshold}')
if 'snapshot_length' in rule_conf['log_options']:
log_snaplen = rule_conf['log_options']['snapshot_length']
output.append(f'snaplen {log_snaplen}')
output.append('counter')
if 'set' in rule_conf:
output.append(parse_policy_set(rule_conf['set'], def_suffix))
if 'action' in rule_conf:
# Change action=return to action=action
# #output.append(nft_action(rule_conf['action']))
if rule_conf['action'] == 'offload':
offload_target = rule_conf['offload_target']
output.append(f'flow add @VYOS_FLOWTABLE_{offload_target}')
else:
output.append(f'{rule_conf["action"]}')
if 'jump' in rule_conf['action']:
target = rule_conf['jump_target']
output.append(f'NAME{def_suffix}_{target}')
if 'queue' in rule_conf['action']:
if 'queue' in rule_conf:
target = rule_conf['queue']
output.append(f'num {target}')
if 'queue_options' in rule_conf:
queue_opts = ','.join(rule_conf['queue_options'])
output.append(f'{queue_opts}')
# Synproxy
if 'synproxy' in rule_conf:
synproxy_mss = dict_search_args(rule_conf, 'synproxy', 'tcp', 'mss')
if synproxy_mss:
output.append(f'mss {synproxy_mss}')
synproxy_ws = dict_search_args(rule_conf, 'synproxy', 'tcp', 'window_scale')
if synproxy_ws:
output.append(f'wscale {synproxy_ws} timestamp sack-perm')
else:
output.append('return')
output.append(f'comment "{family}-{hook}-{fw_name}-{rule_id}"')
return " ".join(output)
def parse_tcp_flags(flags):
include = [flag for flag in flags if flag != 'not']
exclude = list(flags['not']) if 'not' in flags else []
return f'tcp flags & ({"|".join(include + exclude)}) == {"|".join(include) if include else "0x0"}'
def parse_time(time):
out = []
if 'startdate' in time:
start = time['startdate']
if 'T' not in start and 'starttime' in time:
start += f' {time["starttime"]}'
out.append(f'time >= "{start}"')
if 'starttime' in time and 'startdate' not in time:
out.append(f'hour >= "{time["starttime"]}"')
if 'stopdate' in time:
stop = time['stopdate']
if 'T' not in stop and 'stoptime' in time:
stop += f' {time["stoptime"]}'
out.append(f'time < "{stop}"')
if 'stoptime' in time and 'stopdate' not in time:
out.append(f'hour < "{time["stoptime"]}"')
if 'weekdays' in time:
days = time['weekdays'].split(",")
out_days = [f'"{day}"' for day in days if day[0] != '!']
out.append(f'day {{{",".join(out_days)}}}')
return " ".join(out)
def parse_policy_set(set_conf, def_suffix):
out = []
if 'connection_mark' in set_conf:
conn_mark = set_conf['connection_mark']
out.append(f'ct mark set {conn_mark}')
if 'dscp' in set_conf:
dscp = set_conf['dscp']
out.append(f'ip{def_suffix} dscp set {dscp}')
if 'mark' in set_conf:
mark = set_conf['mark']
out.append(f'meta mark set {mark}')
if 'table' in set_conf:
table = set_conf['table']
if table == 'main':
table = '254'
mark = 0x7FFFFFFF - int(table)
out.append(f'meta mark set {mark}')
if 'tcp_mss' in set_conf:
mss = set_conf['tcp_mss']
out.append(f'tcp option maxseg size set {mss}')
return " ".join(out)
# GeoIP
nftables_geoip_conf = '/run/nftables-geoip.conf'
geoip_database = '/usr/share/vyos-geoip/dbip-country-lite.csv.gz'
geoip_lock_file = '/run/vyos-geoip.lock'
def geoip_load_data(codes=[]):
data = None
if not os.path.exists(geoip_database):
return []
try:
with gzip.open(geoip_database, mode='rt') as csv_fh:
reader = csv.reader(csv_fh)
out = []
for start, end, code in reader:
if code.lower() in codes:
out.append([start, end, code.lower()])
return out
except:
print('Error: Failed to open GeoIP database')
return []
def geoip_download_data():
url = 'https://download.db-ip.com/free/dbip-country-lite-{}.csv.gz'.format(strftime("%Y-%m"))
try:
dirname = os.path.dirname(geoip_database)
if not os.path.exists(dirname):
os.mkdir(dirname)
download(geoip_database, url)
print("Downloaded GeoIP database")
return True
except:
print("Error: Failed to download GeoIP database")
return False
class GeoIPLock(object):
def __init__(self, file):
self.file = file
def __enter__(self):
if os.path.exists(self.file):
return False
Path(self.file).touch()
return True
def __exit__(self, exc_type, exc_value, tb):
os.unlink(self.file)
def geoip_update(firewall, force=False):
with GeoIPLock(geoip_lock_file) as lock:
if not lock:
print("Script is already running")
return False
if not firewall:
print("Firewall is not configured")
return True
if not os.path.exists(geoip_database):
if not geoip_download_data():
return False
elif force:
geoip_download_data()
ipv4_codes = {}
ipv6_codes = {}
ipv4_sets = {}
ipv6_sets = {}
# Map country codes to set names
for codes, path in dict_search_recursive(firewall, 'country_code'):
set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}'
if ( path[0] == 'ipv4'):
for code in codes:
ipv4_codes.setdefault(code, []).append(set_name)
elif ( path[0] == 'ipv6' ):
set_name = f'GEOIP_CC6_{path[1]}_{path[2]}_{path[4]}'
for code in codes:
ipv6_codes.setdefault(code, []).append(set_name)
if not ipv4_codes and not ipv6_codes:
if force:
print("GeoIP not in use by firewall")
return True
geoip_data = geoip_load_data([*ipv4_codes, *ipv6_codes])
# Iterate IP blocks to assign to sets
for start, end, code in geoip_data:
ipv4 = is_ipv4(start)
if code in ipv4_codes and ipv4:
ip_range = f'{start}-{end}' if start != end else start
for setname in ipv4_codes[code]:
ipv4_sets.setdefault(setname, []).append(ip_range)
if code in ipv6_codes and not ipv4:
ip_range = f'{start}-{end}' if start != end else start
for setname in ipv6_codes[code]:
ipv6_sets.setdefault(setname, []).append(ip_range)
render(nftables_geoip_conf, 'firewall/nftables-geoip-update.j2', {
'ipv4_sets': ipv4_sets,
'ipv6_sets': ipv6_sets
})
result = run(f'nft -f {nftables_geoip_conf}')
if result != 0:
print('Error: GeoIP failed to update firewall')
return False
return True
diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py
index 67e949f95..7b4ba11d0 100755
--- a/smoketest/scripts/cli/test_firewall.py
+++ b/smoketest/scripts/cli/test_firewall.py
@@ -1,674 +1,677 @@
#!/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 unittest
from glob import glob
from time import sleep
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.configsession import ConfigSessionError
from vyos.utils.process import cmd
from vyos.utils.process import run
sysfs_config = {
'all_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_all', 'default': '0', 'test_value': 'disable'},
'broadcast_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_broadcasts', 'default': '1', 'test_value': 'enable'},
'ip_src_route': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_source_route', 'default': '0', 'test_value': 'enable'},
'ipv6_receive_redirects': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_redirects', 'default': '0', 'test_value': 'enable'},
'ipv6_src_route': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_source_route', 'default': '-1', 'test_value': 'enable'},
'log_martians': {'sysfs': '/proc/sys/net/ipv4/conf/all/log_martians', 'default': '1', 'test_value': 'disable'},
'receive_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_redirects', 'default': '0', 'test_value': 'enable'},
'send_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/send_redirects', 'default': '1', 'test_value': 'disable'},
'syn_cookies': {'sysfs': '/proc/sys/net/ipv4/tcp_syncookies', 'default': '1', 'test_value': 'disable'},
'twa_hazards_protection': {'sysfs': '/proc/sys/net/ipv4/tcp_rfc1337', 'default': '0', 'test_value': 'enable'}
}
class TestFirewall(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestFirewall, cls).setUpClass()
# ensure we can also run this test on a live system - so lets clean
# out the current configuration :)
cls.cli_delete(cls, ['firewall'])
@classmethod
def tearDownClass(cls):
super(TestFirewall, cls).tearDownClass()
def tearDown(self):
self.cli_delete(['firewall'])
self.cli_commit()
# Verify chains/sets are cleaned up from nftables
nftables_search = [
['set M_smoketest_mac'],
['set N_smoketest_network'],
['set P_smoketest_port'],
['set D_smoketest_domain'],
['set RECENT_smoketest_4'],
['chain NAME_smoketest']
]
self.verify_nftables(nftables_search, 'ip vyos_filter', inverse=True)
def verify_nftables(self, nftables_search, table, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list table {table}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def verify_nftables_chain(self, nftables_search, table, chain, inverse=False, args=''):
nftables_output = cmd(f'sudo nft {args} list chain {table} {chain}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def wait_for_domain_resolver(self, table, set_name, element, max_wait=10):
# Resolver no longer blocks commit, need to wait for daemon to populate set
count = 0
while count < max_wait:
code = run(f'sudo nft get element {table} {set_name} {{ {element} }}')
if code == 0:
return True
count += 1
sleep(1)
return False
def test_geoip(self):
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'source', 'geoip', 'country-code', 'se'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'source', 'geoip', 'country-code', 'gb'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '2', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '2', 'source', 'geoip', 'country-code', 'de'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '2', 'source', 'geoip', 'country-code', 'fr'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '2', 'source', 'geoip', 'inverse-match'])
self.cli_commit()
nftables_search = [
['ip saddr @GEOIP_CC_name_smoketest_1', 'drop'],
['ip saddr != @GEOIP_CC_name_smoketest_2', 'accept']
]
# -t prevents 1000+ GeoIP elements being returned
self.verify_nftables(nftables_search, 'ip vyos_filter', args='-t')
def test_groups(self):
hostmap_path = ['system', 'static-host-mapping', 'host-name']
example_org = ['192.0.2.8', '192.0.2.10', '192.0.2.11']
self.cli_set(hostmap_path + ['example.com', 'inet', '192.0.2.5'])
for ips in example_org:
self.cli_set(hostmap_path + ['example.org', 'inet', ips])
self.cli_commit()
self.cli_set(['firewall', 'group', 'mac-group', 'smoketest_mac', 'mac-address', '00:01:02:03:04:05'])
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network', 'network', '172.16.99.0/24'])
self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port', 'port', '53'])
self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port', 'port', '123'])
self.cli_set(['firewall', 'group', 'domain-group', 'smoketest_domain', 'address', 'example.com'])
self.cli_set(['firewall', 'group', 'domain-group', 'smoketest_domain', 'address', 'example.org'])
self.cli_set(['firewall', 'group', 'interface-group', 'smoketest_interface', 'interface', 'eth0'])
self.cli_set(['firewall', 'group', 'interface-group', 'smoketest_interface', 'interface', 'vtun0'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'source', 'group', 'network-group', 'smoketest_network'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'destination', 'address', '172.16.10.10'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'destination', 'group', 'port-group', 'smoketest_port'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'protocol', 'tcp_udp'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '2', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '2', 'source', 'group', 'mac-group', 'smoketest_mac'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'source', 'group', 'domain-group', 'smoketest_domain'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'outbound-interface', 'interface-group', '!smoketest_interface'])
self.cli_commit()
self.wait_for_domain_resolver('ip vyos_filter', 'D_smoketest_domain', '192.0.2.5')
nftables_search = [
['ip saddr @N_smoketest_network', 'ip daddr 172.16.10.10', 'th dport @P_smoketest_port', 'accept'],
['elements = { 172.16.99.0/24 }'],
['elements = { 53, 123 }'],
['ether saddr @M_smoketest_mac', 'accept'],
['elements = { 00:01:02:03:04:05 }'],
['set D_smoketest_domain'],
['elements = { 192.0.2.5, 192.0.2.8,'],
['192.0.2.10, 192.0.2.11 }'],
['ip saddr @D_smoketest_domain', 'accept'],
['oifname != @I_smoketest_interface', 'accept']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
self.cli_delete(['system', 'static-host-mapping'])
self.cli_commit()
def test_nested_groups(self):
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network', 'network', '172.16.99.0/24'])
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network1', 'network', '172.16.101.0/24'])
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network1', 'include', 'smoketest_network'])
self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port', 'port', '53'])
self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port1', 'port', '123'])
self.cli_set(['firewall', 'group', 'port-group', 'smoketest_port1', 'include', 'smoketest_port'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'source', 'group', 'network-group', 'smoketest_network1'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'destination', 'group', 'port-group', 'smoketest_port1'])
self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'protocol', 'tcp_udp'])
self.cli_commit()
# Test circular includes
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network', 'include', 'smoketest_network1'])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_delete(['firewall', 'group', 'network-group', 'smoketest_network', 'include', 'smoketest_network1'])
nftables_search = [
['ip saddr @N_smoketest_network1', 'th dport @P_smoketest_port1', 'accept'],
['elements = { 172.16.99.0/24, 172.16.101.0/24 }'],
['elements = { 53, 123 }']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
def test_ipv4_basic_rules(self):
name = 'smoketest'
interface = 'eth0'
interface_inv = '!eth0'
interface_wc = 'l2tp*'
mss_range = '501-1460'
conn_mark = '555'
self.cli_set(['firewall', 'ipv4', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'source', 'address', '172.16.20.10'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'destination', 'address', '172.16.10.10'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'log', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'log-options', 'level', 'debug'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'ttl', 'eq', '15'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'action', 'reject'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'protocol', 'tcp'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'destination', 'port', '8888'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'log', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'log-options', 'level', 'err'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'tcp', 'flags', 'syn'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'tcp', 'flags', 'not', 'ack'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'ttl', 'gt', '102'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'protocol', 'tcp'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'destination', 'port', '22'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'limit', 'rate', '5/minute'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '3', 'log', 'disable'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'protocol', 'tcp'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'destination', 'port', '22'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'recent', 'count', '10'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'recent', 'time', 'minute'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '4', 'packet-type', 'host'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'protocol', 'tcp'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'tcp', 'flags', 'syn'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'tcp', 'mss', mss_range])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'packet-type', 'broadcast'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '5', 'inbound-interface', 'interface-name', interface_wc])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '6', 'action', 'return'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '6', 'protocol', 'gre'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '6', 'connection-mark', conn_mark])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'default-action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '5', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '5', 'protocol', 'gre'])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '5', 'outbound-interface', 'interface-name', interface_inv])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '6', 'action', 'return'])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '6', 'protocol', 'icmp'])
self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'rule', '6', 'connection-mark', conn_mark])
self.cli_commit()
mark_hex = "{0:#010x}".format(int(conn_mark))
nftables_search = [
['chain VYOS_FORWARD_filter'],
['type filter hook forward priority filter; policy drop;'],
['tcp dport 22', 'limit rate 5/minute', 'accept'],
['tcp dport 22', 'add @RECENT_FWD_filter_4 { ip saddr limit rate over 10/minute burst 10 packets }', 'meta pkttype host', 'drop'],
['chain VYOS_INPUT_filter'],
['type filter hook input priority filter; policy accept;'],
['tcp flags & syn == syn', f'tcp option maxseg size {mss_range}', f'iifname "{interface_wc}"', 'meta pkttype broadcast', 'accept'],
['meta l4proto gre', f'ct mark {mark_hex}', 'return'],
['chain VYOS_OUTPUT_filter'],
['type filter hook output priority filter; policy accept;'],
['meta l4proto gre', f'oifname != "{interface}"', 'drop'],
['meta l4proto icmp', f'ct mark {mark_hex}', 'return'],
['chain NAME_smoketest'],
['saddr 172.16.20.10', 'daddr 172.16.10.10', 'log prefix "[ipv4-NAM-smoketest-1-A]" log level debug', 'ip ttl 15', 'accept'],
['tcp flags syn / syn,ack', 'tcp dport 8888', 'log prefix "[ipv4-NAM-smoketest-2-R]" log level err', 'ip ttl > 102', 'reject'],
['log prefix "[smoketest-default-D]"','smoketest default-action', 'drop']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
def test_ipv4_advanced(self):
name = 'smoketest-adv'
name2 = 'smoketest-adv2'
interface = 'eth0'
self.cli_set(['firewall', 'ipv4', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'packet-length', '64'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'packet-length', '512'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'packet-length', '1024'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'dscp', '17'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'dscp', '52'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'log', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'log-options', 'group', '66'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'log-options', 'snapshot-length', '6666'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '6', 'log-options', 'queue-threshold','32000'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '7', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '7', 'packet-length', '1-30000'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '7', 'packet-length-exclude', '60000-65535'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '7', 'dscp', '3-11'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '7', 'dscp-exclude', '21-25'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'source', 'address', '198.51.100.1'])
+ self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'mark', '1010'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'action', 'jump'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'jump-target', name])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '2', 'protocol', 'tcp'])
+ self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '2', 'mark', '!98765'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '2', 'action', 'queue'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '2', 'queue', '3'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '3', 'protocol', 'udp'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '3', 'action', 'queue'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '3', 'queue-options', 'fanout'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '3', 'queue-options', 'bypass'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '3', 'queue', '0-15'])
self.cli_commit()
nftables_search = [
['chain VYOS_FORWARD_filter'],
['type filter hook forward priority filter; policy drop;'],
- ['ip saddr 198.51.100.1', f'jump NAME_{name}'],
+ ['ip saddr 198.51.100.1', 'meta mark 0x000003f2', f'jump NAME_{name}'],
['chain VYOS_INPUT_filter'],
['type filter hook input priority filter; policy accept;'],
- [f'meta l4proto tcp','queue to 3'],
- [f'meta l4proto udp','queue flags bypass,fanout to 0-15'],
+ ['meta mark != 0x000181cd', 'meta l4proto tcp','queue to 3'],
+ ['meta l4proto udp','queue flags bypass,fanout to 0-15'],
[f'chain NAME_{name}'],
['ip length { 64, 512, 1024 }', 'ip dscp { 0x11, 0x34 }', f'log prefix "[ipv4-NAM-{name}-6-A]" log group 66 snaplen 6666 queue-threshold 32000', 'accept'],
['ip length 1-30000', 'ip length != 60000-65535', 'ip dscp 0x03-0x0b', 'ip dscp != 0x15-0x19', 'accept'],
[f'log prefix "[{name}-default-D]"', 'drop']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
def test_ipv4_synproxy(self):
tcp_mss = '1460'
tcp_wscale = '7'
dport = '22'
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'protocol', 'tcp'])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'destination', 'port', dport])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'synproxy', 'tcp', 'mss', tcp_mss])
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'synproxy', 'tcp', 'window-scale', tcp_wscale])
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'action', 'synproxy'])
self.cli_commit()
nftables_search = [
[f'tcp dport {dport} ct state invalid,untracked', f'synproxy mss {tcp_mss} wscale {tcp_wscale} timestamp sack-perm']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
def test_ipv4_mask(self):
name = 'smoketest-mask'
interface = 'eth0'
self.cli_set(['firewall', 'group', 'address-group', 'mask_group', 'address', '1.1.1.1'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'destination', 'address', '0.0.1.2'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'destination', 'address-mask', '0.0.255.255'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'source', 'address', '!0.0.3.4'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'source', 'address-mask', '0.0.255.255'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'source', 'group', 'address-group', 'mask_group'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'source', 'address-mask', '0.0.255.255'])
self.cli_commit()
nftables_search = [
[f'daddr & 0.0.255.255 == 0.0.1.2'],
[f'saddr & 0.0.255.255 != 0.0.3.4'],
[f'saddr & 0.0.255.255 == @A_mask_group']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
def test_ipv6_basic_rules(self):
name = 'v6-smoketest'
interface = 'eth0'
self.cli_set(['firewall', 'ipv6', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'source', 'address', '2002::1'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'destination', 'address', '2002::1:1'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'log', 'enable'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'log-options', 'level', 'crit'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'default-action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '2', 'action', 'reject'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '2', 'protocol', 'tcp_udp'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '2', 'destination', 'port', '8888'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '2', 'inbound-interface', 'interface-name', interface])
self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'rule', '3', 'action', 'return'])
self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'rule', '3', 'protocol', 'gre'])
self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'rule', '3', 'outbound-interface', 'interface-name', interface])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '3', 'action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '3', 'protocol', 'udp'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '3', 'source', 'address', '2002::1:2'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '3', 'inbound-interface', 'interface-name', interface])
self.cli_commit()
nftables_search = [
['chain VYOS_IPV6_FORWARD_filter'],
['type filter hook forward priority filter; policy accept;'],
['meta l4proto { tcp, udp }', 'th dport 8888', f'iifname "{interface}"', 'reject'],
['chain VYOS_IPV6_INPUT_filter'],
['type filter hook input priority filter; policy accept;'],
['meta l4proto udp', 'ip6 saddr 2002::1:2', f'iifname "{interface}"', 'accept'],
['chain VYOS_IPV6_OUTPUT_filter'],
['type filter hook output priority filter; policy drop;'],
['meta l4proto gre', f'oifname "{interface}"', 'return'],
[f'chain NAME6_{name}'],
['saddr 2002::1', 'daddr 2002::1:1', 'log prefix "[ipv6-NAM-v6-smoketest-1-A]" log level crit', 'accept'],
[f'"{name} default-action drop"', f'log prefix "[{name}-default-D]"', 'drop']
]
self.verify_nftables(nftables_search, 'ip6 vyos_filter')
def test_ipv6_advanced(self):
name = 'v6-smoketest-adv'
name2 = 'v6-smoketest-adv2'
interface = 'eth0'
self.cli_set(['firewall', 'ipv6', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'packet-length', '65'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'packet-length', '513'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'packet-length', '1025'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'dscp', '18'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'dscp', '53'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '4', 'action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '4', 'packet-length', '1-1999'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '4', 'packet-length-exclude', '60000-65535'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '4', 'dscp', '4-14'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '4', 'dscp-exclude', '31-35'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'default-action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '1', 'source', 'address', '2001:db8::/64'])
+ self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '1', 'mark', '!6655-7766'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '1', 'action', 'jump'])
self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '1', 'jump-target', name])
self.cli_commit()
nftables_search = [
['chain VYOS_IPV6_FORWARD_filter'],
['type filter hook forward priority filter; policy accept;'],
['ip6 length 1-1999', 'ip6 length != 60000-65535', 'ip6 dscp 0x04-0x0e', 'ip6 dscp != 0x1f-0x23', 'accept'],
['chain VYOS_IPV6_INPUT_filter'],
['type filter hook input priority filter; policy accept;'],
- ['ip6 saddr 2001:db8::/64', f'jump NAME6_{name}'],
+ ['ip6 saddr 2001:db8::/64', 'meta mark != 0x000019ff-0x00001e56', f'jump NAME6_{name}'],
[f'chain NAME6_{name}'],
['ip6 length { 65, 513, 1025 }', 'ip6 dscp { af21, 0x35 }', 'accept'],
[f'log prefix "[{name}-default-D]"', 'drop']
]
self.verify_nftables(nftables_search, 'ip6 vyos_filter')
def test_ipv6_mask(self):
name = 'v6-smoketest-mask'
interface = 'eth0'
self.cli_set(['firewall', 'group', 'ipv6-address-group', 'mask_group', 'address', '::beef'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'destination', 'address', '::1111:2222:3333:4444'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '1', 'destination', 'address-mask', '::ffff:ffff:ffff:ffff'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '2', 'action', 'accept'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '2', 'source', 'address', '!::aaaa:bbbb:cccc:dddd'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '2', 'source', 'address-mask', '::ffff:ffff:ffff:ffff'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'action', 'drop'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'source', 'group', 'address-group', 'mask_group'])
self.cli_set(['firewall', 'ipv6', 'name', name, 'rule', '3', 'source', 'address-mask', '::ffff:ffff:ffff:ffff'])
self.cli_commit()
nftables_search = [
['daddr & ::ffff:ffff:ffff:ffff == ::1111:2222:3333:4444'],
['saddr & ::ffff:ffff:ffff:ffff != ::aaaa:bbbb:cccc:dddd'],
['saddr & ::ffff:ffff:ffff:ffff == @A6_mask_group']
]
self.verify_nftables(nftables_search, 'ip6 vyos_filter')
def test_ipv4_state_and_status_rules(self):
name = 'smoketest-state'
interface = 'eth0'
self.cli_set(['firewall', 'ipv4', 'name', name, 'default-action', 'drop'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'state', 'established', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'state', 'related', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'action', 'reject'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'state', 'invalid', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'state', 'new', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'connection-status', 'nat', 'destination'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '4', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '4', 'state', 'new', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '4', 'state', 'established', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '4', 'connection-status', 'nat', 'source'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '5', 'action', 'accept'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '5', 'state', 'related', 'enable'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '5', 'conntrack-helper', 'ftp'])
self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '5', 'conntrack-helper', 'pptp'])
self.cli_commit()
nftables_search = [
['ct state { established, related }', 'accept'],
['ct state invalid', 'reject'],
['ct state new', 'ct status dnat', 'accept'],
['ct state { established, new }', 'ct status snat', 'accept'],
['ct state related', 'ct helper { "ftp", "pptp" }', 'accept'],
['drop', f'comment "{name} default-action drop"']
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
# Check conntrack
self.verify_nftables_chain([['accept']], 'ip vyos_conntrack', 'FW_CONNTRACK')
self.verify_nftables_chain([['return']], 'ip6 vyos_conntrack', 'FW_CONNTRACK')
def test_bridge_basic_rules(self):
name = 'smoketest'
interface_in = 'eth0'
mac_address = '00:53:00:00:00:01'
vlan_id = '12'
vlan_prior = '3'
self.cli_set(['firewall', 'bridge', 'name', name, 'default-action', 'accept'])
self.cli_set(['firewall', 'bridge', 'name', name, 'enable-default-log'])
self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'source', 'mac-address', mac_address])
self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'inbound-interface', 'interface-name', interface_in])
self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'log', 'enable'])
self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'log-options', 'level', 'crit'])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'default-action', 'drop'])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'rule', '1', 'action', 'accept'])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'rule', '1', 'vlan', 'id', vlan_id])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'rule', '2', 'action', 'jump'])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'rule', '2', 'jump-target', name])
self.cli_set(['firewall', 'bridge', 'forward', 'filter', 'rule', '2', 'vlan', 'priority', vlan_prior])
self.cli_commit()
nftables_search = [
['chain VYOS_FORWARD_filter'],
['type filter hook forward priority filter; policy drop;'],
[f'vlan id {vlan_id}', 'accept'],
[f'vlan pcp {vlan_prior}', f'jump NAME_{name}'],
[f'chain NAME_{name}'],
[f'ether saddr {mac_address}', f'iifname "{interface_in}"', f'log prefix "[bri-NAM-{name}-1-A]" log level crit', 'accept']
]
self.verify_nftables(nftables_search, 'bridge vyos_filter')
def test_source_validation(self):
# Strict
self.cli_set(['firewall', 'global-options', 'source-validation', 'strict'])
self.cli_set(['firewall', 'global-options', 'ipv6-source-validation', 'strict'])
self.cli_commit()
nftables_strict_search = [
['fib saddr . iif oif 0', 'drop']
]
self.verify_nftables_chain(nftables_strict_search, 'ip raw', 'vyos_global_rpfilter')
self.verify_nftables_chain(nftables_strict_search, 'ip6 raw', 'vyos_global_rpfilter')
# Loose
self.cli_set(['firewall', 'global-options', 'source-validation', 'loose'])
self.cli_set(['firewall', 'global-options', 'ipv6-source-validation', 'loose'])
self.cli_commit()
nftables_loose_search = [
['fib saddr oif 0', 'drop']
]
self.verify_nftables_chain(nftables_loose_search, 'ip raw', 'vyos_global_rpfilter')
self.verify_nftables_chain(nftables_loose_search, 'ip6 raw', 'vyos_global_rpfilter')
def test_sysfs(self):
for name, conf in sysfs_config.items():
paths = glob(conf['sysfs'])
for path in paths:
with open(path, 'r') as f:
self.assertEqual(f.read().strip(), conf['default'], msg=path)
self.cli_set(['firewall', 'global-options', name.replace("_", "-"), conf['test_value']])
self.cli_commit()
for name, conf in sysfs_config.items():
paths = glob(conf['sysfs'])
for path in paths:
with open(path, 'r') as f:
self.assertNotEqual(f.read().strip(), conf['default'], msg=path)
def test_flow_offload(self):
self.cli_set(['firewall', 'flowtable', 'smoketest', 'interface', 'eth0'])
self.cli_set(['firewall', 'flowtable', 'smoketest', 'offload', 'hardware'])
# QEMU virtual NIC does not support hw-tc-offload
with self.assertRaises(ConfigSessionError):
self.cli_commit()
self.cli_set(['firewall', 'flowtable', 'smoketest', 'offload', 'software'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'action', 'offload'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'offload-target', 'smoketest'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'protocol', 'tcp_udp'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'state', 'established', 'enable'])
self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'state', 'related', 'enable'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'action', 'offload'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'offload-target', 'smoketest'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'protocol', 'tcp_udp'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'state', 'established', 'enable'])
self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'state', 'related', 'enable'])
self.cli_commit()
nftables_search = [
['flowtable VYOS_FLOWTABLE_smoketest'],
['hook ingress priority filter'],
['devices = { eth0 }'],
['ct state { established, related }', 'meta l4proto { tcp, udp }', 'flow add @VYOS_FLOWTABLE_smoketest'],
]
self.verify_nftables(nftables_search, 'ip vyos_filter')
self.verify_nftables(nftables_search, 'ip6 vyos_filter')
# Check conntrack
self.verify_nftables_chain([['accept']], 'ip vyos_conntrack', 'FW_CONNTRACK')
self.verify_nftables_chain([['accept']], 'ip6 vyos_conntrack', 'FW_CONNTRACK')
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/smoketest/scripts/cli/test_policy_route.py b/smoketest/scripts/cli/test_policy_route.py
index c7ddf873e..72192fb98 100755
--- a/smoketest/scripts/cli/test_policy_route.py
+++ b/smoketest/scripts/cli/test_policy_route.py
@@ -1,272 +1,276 @@
#!/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 unittest
from base_vyostest_shim import VyOSUnitTestSHIM
from vyos.utils.process import cmd
mark = '100'
conn_mark = '555'
conn_mark_set = '111'
table_mark_offset = 0x7fffffff
table_id = '101'
interface = 'eth0'
interface_wc = 'ppp*'
interface_ip = '172.16.10.1/24'
class TestPolicyRoute(VyOSUnitTestSHIM.TestCase):
@classmethod
def setUpClass(cls):
super(TestPolicyRoute, cls).setUpClass()
cls.cli_set(cls, ['interfaces', 'ethernet', interface, 'address', interface_ip])
cls.cli_set(cls, ['protocols', 'static', 'table', table_id, 'route', '0.0.0.0/0', 'interface', interface])
@classmethod
def tearDownClass(cls):
cls.cli_delete(cls, ['interfaces', 'ethernet', interface, 'address', interface_ip])
cls.cli_delete(cls, ['protocols', 'static', 'table', table_id])
super(TestPolicyRoute, cls).tearDownClass()
def tearDown(self):
self.cli_delete(['policy', 'route'])
self.cli_delete(['policy', 'route6'])
self.cli_commit()
# Verify nftables cleanup
nftables_search = [
['set N_smoketest_network'],
['set N_smoketest_network1'],
['chain VYOS_PBR_smoketest']
]
self.verify_nftables(nftables_search, 'ip vyos_mangle', inverse=True)
# Verify ip rule cleanup
ip_rule_search = [
['fwmark ' + hex(table_mark_offset - int(table_id)), 'lookup ' + table_id]
]
self.verify_rules(ip_rule_search, inverse=True)
def verify_nftables(self, nftables_search, table, inverse=False):
nftables_output = cmd(f'sudo nft list table {table}')
for search in nftables_search:
matched = False
for line in nftables_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def verify_rules(self, rules_search, inverse=False):
rule_output = cmd('ip rule show')
for search in rules_search:
matched = False
for line in rule_output.split("\n"):
if all(item in line for item in search):
matched = True
break
self.assertTrue(not matched if inverse else matched, msg=search)
def test_pbr_group(self):
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network', 'network', '172.16.99.0/24'])
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network1', 'network', '172.16.101.0/24'])
self.cli_set(['firewall', 'group', 'network-group', 'smoketest_network1', 'include', 'smoketest_network'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'source', 'group', 'network-group', 'smoketest_network'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'destination', 'group', 'network-group', 'smoketest_network1'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'set', 'mark', mark])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface])
self.cli_commit()
nftables_search = [
[f'iifname "{interface}"','jump VYOS_PBR_UD_smoketest'],
['ip daddr @N_smoketest_network1', 'ip saddr @N_smoketest_network'],
]
self.verify_nftables(nftables_search, 'ip vyos_mangle')
self.cli_delete(['firewall'])
def test_pbr_mark(self):
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'source', 'address', '172.16.20.10'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'destination', 'address', '172.16.10.10'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'set', 'mark', mark])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface])
self.cli_commit()
mark_hex = "{0:#010x}".format(int(mark))
nftables_search = [
[f'iifname "{interface}"','jump VYOS_PBR_UD_smoketest'],
['ip daddr 172.16.10.10', 'ip saddr 172.16.20.10', 'meta mark set ' + mark_hex],
]
self.verify_nftables(nftables_search, 'ip vyos_mangle')
def test_pbr_mark_connection(self):
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'source', 'address', '172.16.20.10'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'destination', 'address', '172.16.10.10'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'connection-mark', conn_mark])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'set', 'connection-mark', conn_mark_set])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface])
self.cli_commit()
mark_hex = "{0:#010x}".format(int(conn_mark))
mark_hex_set = "{0:#010x}".format(int(conn_mark_set))
nftables_search = [
[f'iifname "{interface}"','jump VYOS_PBR_UD_smoketest'],
['ip daddr 172.16.10.10', 'ip saddr 172.16.20.10', 'ct mark ' + mark_hex, 'ct mark set ' + mark_hex_set],
]
self.verify_nftables(nftables_search, 'ip vyos_mangle')
def test_pbr_table(self):
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'protocol', 'tcp'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'destination', 'port', '8888'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'tcp', 'flags', 'syn'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'tcp', 'flags', 'not', 'ack'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'set', 'table', table_id])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'protocol', 'tcp_udp'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'destination', 'port', '8888'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'set', 'table', table_id])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface])
self.cli_set(['policy', 'route6', 'smoketest6', 'interface', interface])
self.cli_commit()
mark_hex = "{0:#010x}".format(table_mark_offset - int(table_id))
# IPv4
nftables_search = [
[f'iifname "{interface}"', 'jump VYOS_PBR_UD_smoketest'],
['tcp flags syn / syn,ack', 'tcp dport 8888', 'meta mark set ' + mark_hex]
]
self.verify_nftables(nftables_search, 'ip vyos_mangle')
# IPv6
nftables6_search = [
[f'iifname "{interface}"', 'jump VYOS_PBR6_UD_smoketest'],
['meta l4proto { tcp, udp }', 'th dport 8888', 'meta mark set ' + mark_hex]
]
self.verify_nftables(nftables6_search, 'ip6 vyos_mangle')
# IP rule fwmark -> table
ip_rule_search = [
['fwmark ' + hex(table_mark_offset - int(table_id)), 'lookup ' + table_id]
]
self.verify_rules(ip_rule_search)
def test_pbr_matching_criteria(self):
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'protocol', 'udp'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'action', 'drop'])
+ self.cli_set(['policy', 'route', 'smoketest', 'rule', '1', 'mark', '2020'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'protocol', 'tcp'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'tcp', 'flags', 'syn'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'tcp', 'flags', 'not', 'ack'])
+ self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'mark', '2-3000'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '2', 'set', 'table', table_id])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'source', 'address', '198.51.100.0/24'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'protocol', 'tcp'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'destination', 'port', '22'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'state', 'new', 'enable'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'ttl', 'gt', '2'])
+ self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'mark', '!456'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '3', 'set', 'table', table_id])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'protocol', 'icmp'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'icmp', 'type-name', 'echo-request'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'packet-length', '128'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'packet-length', '1024-2048'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'packet-type', 'other'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'log', 'enable'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '4', 'set', 'table', table_id])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'dscp', '41'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'dscp', '57-59'])
+ self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'mark', '!456-500'])
self.cli_set(['policy', 'route', 'smoketest', 'rule', '5', 'set', 'table', table_id])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'protocol', 'udp'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '1', 'action', 'drop'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'protocol', 'tcp'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'tcp', 'flags', 'syn'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'tcp', 'flags', 'not', 'ack'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '2', 'set', 'table', table_id])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'source', 'address', '2001:db8::0/64'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'protocol', 'tcp'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'destination', 'port', '22'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'state', 'new', 'enable'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'hop-limit', 'gt', '2'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '3', 'set', 'table', table_id])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'protocol', 'icmpv6'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'icmpv6', 'type', 'echo-request'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'packet-length-exclude', '128'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'packet-length-exclude', '1024-2048'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'packet-type', 'multicast'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'log', 'enable'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '4', 'set', 'table', table_id])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'dscp-exclude', '61'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'dscp-exclude', '14-19'])
self.cli_set(['policy', 'route6', 'smoketest6', 'rule', '5', 'set', 'table', table_id])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface])
self.cli_set(['policy', 'route', 'smoketest', 'interface', interface_wc])
self.cli_set(['policy', 'route6', 'smoketest6', 'interface', interface_wc])
self.cli_commit()
mark_hex = "{0:#010x}".format(table_mark_offset - int(table_id))
# IPv4
nftables_search = [
['iifname { "' + interface + '", "' + interface_wc + '" }', 'jump VYOS_PBR_UD_smoketest'],
- ['meta l4proto udp', 'drop'],
- ['tcp flags syn / syn,ack', 'meta mark set ' + mark_hex],
- ['ct state new', 'tcp dport 22', 'ip saddr 198.51.100.0/24', 'ip ttl > 2', 'meta mark set ' + mark_hex],
+ ['meta l4proto udp', 'meta mark 0x000007e4', 'drop'],
+ ['tcp flags syn / syn,ack', 'meta mark 0x00000002-0x00000bb8', 'meta mark set ' + mark_hex],
+ ['ct state new', 'tcp dport 22', 'ip saddr 198.51.100.0/24', 'ip ttl > 2', 'meta mark != 0x000001c8', 'meta mark set ' + mark_hex],
['log prefix "[ipv4-route-smoketest-4-A]"', 'icmp type echo-request', 'ip length { 128, 1024-2048 }', 'meta pkttype other', 'meta mark set ' + mark_hex],
- ['ip dscp { 0x29, 0x39-0x3b }', 'meta mark set ' + mark_hex]
+ ['ip dscp { 0x29, 0x39-0x3b }', 'meta mark != 0x000001c8-0x000001f4', 'meta mark set ' + mark_hex]
]
self.verify_nftables(nftables_search, 'ip vyos_mangle')
# IPv6
nftables6_search = [
[f'iifname "{interface_wc}"', 'jump VYOS_PBR6_UD_smoketest'],
['meta l4proto udp', 'drop'],
['tcp flags syn / syn,ack', 'meta mark set ' + mark_hex],
['ct state new', 'tcp dport 22', 'ip6 saddr 2001:db8::/64', 'ip6 hoplimit > 2', 'meta mark set ' + mark_hex],
['log prefix "[ipv6-route6-smoketest6-4-A]"', 'icmpv6 type echo-request', 'ip6 length != { 128, 1024-2048 }', 'meta pkttype multicast', 'meta mark set ' + mark_hex],
['ip6 dscp != { 0x0e-0x13, 0x3d }', 'meta mark set ' + mark_hex]
]
self.verify_nftables(nftables6_search, 'ip6 vyos_mangle')
if __name__ == '__main__':
unittest.main(verbosity=2)
diff --git a/src/validators/numeric-exclude b/src/validators/numeric-exclude
new file mode 100644
index 000000000..676a240b6
--- /dev/null
+++ b/src/validators/numeric-exclude
@@ -0,0 +1,8 @@
+#!/bin/sh
+path=$(dirname "$0")
+num="${@: -1}"
+if [ "${num:0:1}" != "!" ]; then
+ ${path}/numeric $@
+else
+ ${path}/numeric ${@:1:$#-1} ${num:1}
+fi