diff --git a/interface-definitions/include/firewall/global-options.xml.i b/interface-definitions/include/firewall/global-options.xml.i index 415d85f05..9cd0b3239 100644 --- a/interface-definitions/include/firewall/global-options.xml.i +++ b/interface-definitions/include/firewall/global-options.xml.i @@ -1,313 +1,333 @@ <!-- include start from firewall/global-options.xml.i --> <node name="global-options"> <properties> <help>Global Options</help> </properties> <children> <leafNode name="all-ping"> <properties> <help>Policy for handling of all IPv4 ICMP echo requests</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of all IPv4 ICMP echo requests</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of all IPv4 ICMP echo requests</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>enable</defaultValue> </leafNode> <leafNode name="broadcast-ping"> <properties> <help>Policy for handling broadcast IPv4 ICMP echo and timestamp requests</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of broadcast IPv4 ICMP echo/timestamp requests</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of broadcast IPv4 ICMP echo/timestamp requests</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> + <leafNode name="directed-broadcast"> + <properties> + <help>Policy for handling IPv4 directed broadcast forwarding on all interfaces</help> + <completionHelp> + <list>enable disable</list> + </completionHelp> + <valueHelp> + <format>enable</format> + <description>Enable IPv4 directed broadcast forwarding on all interfaces</description> + </valueHelp> + <valueHelp> + <format>disable</format> + <description>Disable IPv4 directed broadcast forwarding on all interfaces</description> + </valueHelp> + <constraint> + <regex>(enable|disable)</regex> + </constraint> + </properties> + <defaultValue>enable</defaultValue> + </leafNode> <leafNode name="ip-src-route"> <properties> <help>Policy for handling IPv4 packets with source route option</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of IPv4 packets with source route option</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of IPv4 packets with source route option</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <leafNode name="log-martians"> <properties> <help>Policy for logging IPv4 packets with invalid addresses</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable logging of IPv4 packets with invalid addresses</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable logging of Ipv4 packets with invalid addresses</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>enable</defaultValue> </leafNode> <leafNode name="receive-redirects"> <properties> <help>Policy for handling received IPv4 ICMP redirect messages</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of received IPv4 ICMP redirect messages</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of received IPv4 ICMP redirect messages</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <leafNode name="resolver-cache"> <properties> <help>Retains last successful value if domain resolution fails</help> <valueless/> </properties> </leafNode> <leafNode name="resolver-interval"> <properties> <help>Domain resolver update interval</help> <valueHelp> <format>u32:10-3600</format> <description>Interval (seconds)</description> </valueHelp> <constraint> <validator name="numeric" argument="--range 10-3600"/> </constraint> </properties> <defaultValue>300</defaultValue> </leafNode> <leafNode name="send-redirects"> <properties> <help>Policy for sending IPv4 ICMP redirect messages</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable sending IPv4 ICMP redirect messages</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable sending IPv4 ICMP redirect messages</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>enable</defaultValue> </leafNode> <leafNode name="source-validation"> <properties> <help>Policy for IPv4 source validation by reversed path, as specified in RFC3704</help> <completionHelp> <list>strict loose disable</list> </completionHelp> <valueHelp> <format>strict</format> <description>Enable IPv4 Strict Reverse Path Forwarding as defined in RFC3704</description> </valueHelp> <valueHelp> <format>loose</format> <description>Enable IPv4 Loose Reverse Path Forwarding as defined in RFC3704</description> </valueHelp> <valueHelp> <format>disable</format> <description>No IPv4 source validation</description> </valueHelp> <constraint> <regex>(strict|loose|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <node name="state-policy"> <properties> <help>Global firewall state-policy</help> </properties> <children> <node name="established"> <properties> <help>Global firewall policy for packets part of an established connection</help> </properties> <children> #include <include/firewall/action-accept-drop-reject.xml.i> #include <include/firewall/log.xml.i> #include <include/firewall/rule-log-level.xml.i> </children> </node> <node name="invalid"> <properties> <help>Global firewall policy for packets part of an invalid connection</help> </properties> <children> #include <include/firewall/action-accept-drop-reject.xml.i> #include <include/firewall/log.xml.i> #include <include/firewall/rule-log-level.xml.i> </children> </node> <node name="related"> <properties> <help>Global firewall policy for packets part of a related connection</help> </properties> <children> #include <include/firewall/action-accept-drop-reject.xml.i> #include <include/firewall/log.xml.i> #include <include/firewall/rule-log-level.xml.i> </children> </node> </children> </node> <leafNode name="syn-cookies"> <properties> <help>Policy for using TCP SYN cookies with IPv4</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable use of TCP SYN cookies with IPv4</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable use of TCP SYN cookies with IPv4</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>enable</defaultValue> </leafNode> <leafNode name="twa-hazards-protection"> <properties> <help>RFC1337 TCP TIME-WAIT assasination hazards protection</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable RFC1337 TIME-WAIT hazards protection</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable RFC1337 TIME-WAIT hazards protection</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <leafNode name="ipv6-receive-redirects"> <properties> <help>Policy for handling received ICMPv6 redirect messages</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of received ICMPv6 redirect messages</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of received ICMPv6 redirect messages</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <leafNode name="ipv6-source-validation"> <properties> <help>Policy for IPv6 source validation by reversed path, as specified in RFC3704</help> <completionHelp> <list>strict loose disable</list> </completionHelp> <valueHelp> <format>strict</format> <description>Enable IPv6 Strict Reverse Path Forwarding as defined in RFC3704</description> </valueHelp> <valueHelp> <format>loose</format> <description>Enable IPv6 Loose Reverse Path Forwarding as defined in RFC3704</description> </valueHelp> <valueHelp> <format>disable</format> <description>No IPv6 source validation</description> </valueHelp> <constraint> <regex>(strict|loose|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> <leafNode name="ipv6-src-route"> <properties> <help>Policy for handling IPv6 packets with routing extension header</help> <completionHelp> <list>enable disable</list> </completionHelp> <valueHelp> <format>enable</format> <description>Enable processing of IPv6 packets with routing header type 2</description> </valueHelp> <valueHelp> <format>disable</format> <description>Disable processing of IPv6 packets with routing header</description> </valueHelp> <constraint> <regex>(enable|disable)</regex> </constraint> </properties> <defaultValue>disable</defaultValue> </leafNode> </children> </node> <!-- include end --> diff --git a/interface-definitions/include/version/firewall-version.xml.i b/interface-definitions/include/version/firewall-version.xml.i index 6702ee041..fa8e26f78 100644 --- a/interface-definitions/include/version/firewall-version.xml.i +++ b/interface-definitions/include/version/firewall-version.xml.i @@ -1,3 +1,3 @@ <!-- include start from include/version/firewall-version.xml.i --> -<syntaxVersion component='firewall' version='14'></syntaxVersion> +<syntaxVersion component='firewall' version='15'></syntaxVersion> <!-- include end --> diff --git a/interface-definitions/system_ip.xml.in b/interface-definitions/system_ip.xml.in index 015eb270f..b4b5092fe 100644 --- a/interface-definitions/system_ip.xml.in +++ b/interface-definitions/system_ip.xml.in @@ -1,115 +1,109 @@ <?xml version="1.0"?> <interfaceDefinition> <node name="system"> <children> <node name="ip" owner="${vyos_conf_scripts_dir}/system_ip.py"> <properties> <help>IPv4 Settings</help> <!-- must be before any interface, check /opt/vyatta/sbin/priority.pl --> <priority>290</priority> </properties> <children> <node name="arp"> <properties> <help>Parameters for ARP cache</help> </properties> <children> #include <include/arp-ndp-table-size.xml.i> </children> </node> <leafNode name="disable-forwarding"> <properties> <help>Disable IPv4 forwarding on all interfaces</help> <valueless/> </properties> </leafNode> - <leafNode name="disable-directed-broadcast"> - <properties> - <help>Disable IPv4 directed broadcast forwarding on all interfaces</help> - <valueless/> - </properties> - </leafNode> <node name="multipath"> <properties> <help>IPv4 multipath settings</help> </properties> <children> <leafNode name="ignore-unreachable-nexthops"> <properties> <help>Ignore next hops that are not in the ARP table</help> <valueless/> </properties> </leafNode> <leafNode name="layer4-hashing"> <properties> <help>Use layer 4 information for ECMP hashing</help> <valueless/> </properties> </leafNode> </children> </node> #include <include/system-ip-nht.xml.i> <node name="tcp"> <properties> <help>IPv4 TCP parameters</help> </properties> <children> <node name="mss"> <properties> <help>IPv4 TCP MSS probing options</help> </properties> <children> <leafNode name="probing"> <properties> <help>Attempt to lower the MSS if TCP connections fail to establish</help> <completionHelp> <list>on-icmp-black-hole force</list> </completionHelp> <valueHelp> <format>on-icmp-black-hole</format> <description>Attempt TCP MSS probing when an ICMP black hole is detected</description> </valueHelp> <valueHelp> <format>force</format> <description>Attempt TCP MSS probing by default</description> </valueHelp> <constraint> <regex>(on-icmp-black-hole|force)</regex> </constraint> <constraintErrorMessage>Must be on-icmp-black-hole or force</constraintErrorMessage> </properties> </leafNode> <leafNode name="base"> <properties> <help>Base MSS to start probing from (applicable to "probing force")</help> <valueHelp> <format>u32:48-1460</format> <description>Base MSS value for probing (default: 1024)</description> </valueHelp> <constraint> <validator name="numeric" argument="--range 48-1460"/> </constraint> </properties> </leafNode> <leafNode name="floor"> <properties> <help>Minimum MSS to stop probing at (default: 48)</help> <valueHelp> <format>u32:48-1460</format> <description>Minimum MSS value to probe</description> </valueHelp> <constraint> <validator name="numeric" argument="--range 48-1460"/> </constraint> </properties> </leafNode> </children> </node> </children> </node> #include <include/system-ip-protocol.xml.i> </children> </node> </children> </node> </interfaceDefinition> diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py index fe6977252..c47562714 100755 --- a/smoketest/scripts/cli/test_firewall.py +++ b/smoketest/scripts/cli/test_firewall.py @@ -1,878 +1,879 @@ #!/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 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'}, + 'directed_broadcast': {'sysfs': '/proc/sys/net/ipv4/conf/all/bc_forwarding', 'default': '1', 'test_value': 'disable'}, '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 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', '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, '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']) 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']) 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', 'default-log']) 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']) 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', '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', 'drop']) self.cli_set(['firewall', 'ipv4', 'output', 'filter', 'default-log']) 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', '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 accept;'], ['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'], ['log prefix "[ipv4-FWD-filter-default-D]"','FWD-filter default-action drop', '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'], ['INP-filter default-action accept', 'accept'], ['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'], ['log prefix "[ipv4-OUT-filter-default-D]"','OUT-filter default-action drop', 'drop'], ['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 "[ipv4-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, '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']) 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 accept;'], ['ip saddr 198.51.100.1', 'meta mark 0x000003f2', f'jump NAME_{name}'], ['FWD-filter default-action drop', 'drop'], ['chain VYOS_INPUT_filter'], ['type filter hook input priority filter; policy accept;'], ['meta mark != 0x000181cd', 'meta l4proto tcp','queue to 3'], ['meta l4proto udp','queue flags bypass,fanout to 0-15'], ['INP-filter default-action accept', 'accept'], [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 "[ipv4-{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, '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_ipv4_dynamic_groups(self): group01 = 'knock01' group02 = 'allowed' self.cli_set(['firewall', 'group', 'dynamic-group', 'address-group', group01]) self.cli_set(['firewall', 'group', 'dynamic-group', 'address-group', group02]) 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', '5151']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'add-address-to-group', 'source-address', 'address-group', group01]) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '10', 'add-address-to-group', 'source-address', 'timeout', '30s']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'action', 'drop']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'protocol', 'tcp']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'destination', 'port', '7272']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'source', 'group', 'dynamic-address-group', group01]) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'add-address-to-group', 'source-address', 'address-group', group02]) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '20', 'add-address-to-group', 'source-address', 'timeout', '5m']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '30', 'action', 'accept']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '30', 'protocol', 'tcp']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '30', 'destination', 'port', '22']) self.cli_set(['firewall', 'ipv4', 'input', 'filter', 'rule', '30', 'source', 'group', 'dynamic-address-group', group02]) self.cli_commit() nftables_search = [ [f'DA_{group01}'], [f'DA_{group02}'], ['type ipv4_addr'], ['flags dynamic,timeout'], ['chain VYOS_INPUT_filter {'], ['type filter hook input priority filter', 'policy accept'], ['tcp dport 5151', f'update @DA_{group01}', '{ ip saddr timeout 30s }', 'drop'], ['tcp dport 7272', f'ip saddr @DA_{group01}', f'update @DA_{group02}', '{ ip saddr timeout 5m }', 'drop'], ['tcp dport 22', f'ip saddr @DA_{group02}', 'accept'] ] self.verify_nftables(nftables_search, 'ip vyos_filter') def test_ipv6_basic_rules(self): name = 'v6-smoketest' interface = 'eth0' self.cli_set(['firewall', 'global-options', 'state-policy', 'established', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'related', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'invalid', 'action', 'drop']) self.cli_set(['firewall', 'ipv6', 'name', name, 'default-action', 'drop']) self.cli_set(['firewall', 'ipv6', 'name', name, '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']) 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', 'default-log']) 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', 'name', interface]) self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'default-action', 'drop']) self.cli_set(['firewall', 'ipv6', 'output', 'filter', 'default-log']) 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', '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', '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'], ['log prefix "[ipv6-FWD-filter-default-A]"','FWD-filter default-action accept', 'accept'], ['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'], ['INP-filter default-action accept', 'accept'], ['chain VYOS_IPV6_OUTPUT_filter'], ['type filter hook output priority filter; policy accept;'], ['meta l4proto gre', f'oifname "{interface}"', 'return'], ['log prefix "[ipv6-OUT-filter-default-D]"','OUT-filter default-action drop', 'drop'], [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 "[ipv6-{name}-default-D]"', 'drop'], ['jump VYOS_STATE_POLICY6'], ['chain VYOS_STATE_POLICY6'], ['ct state established', 'accept'], ['ct state invalid', 'drop'], ['ct state related', 'accept'] ] 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, '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', '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 "[ipv6-{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, '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_ipv6_dynamic_groups(self): group01 = 'knock01' group02 = 'allowed' self.cli_set(['firewall', 'group', 'dynamic-group', 'ipv6-address-group', group01]) self.cli_set(['firewall', 'group', 'dynamic-group', 'ipv6-address-group', group02]) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '10', 'action', 'drop']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '10', 'protocol', 'tcp']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '10', 'destination', 'port', '5151']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '10', 'add-address-to-group', 'source-address', 'address-group', group01]) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '10', 'add-address-to-group', 'source-address', 'timeout', '30s']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'action', 'drop']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'protocol', 'tcp']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'destination', 'port', '7272']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'source', 'group', 'dynamic-address-group', group01]) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'add-address-to-group', 'source-address', 'address-group', group02]) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '20', 'add-address-to-group', 'source-address', 'timeout', '5m']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '30', 'action', 'accept']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '30', 'protocol', 'tcp']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '30', 'destination', 'port', '22']) self.cli_set(['firewall', 'ipv6', 'input', 'filter', 'rule', '30', 'source', 'group', 'dynamic-address-group', group02]) self.cli_commit() nftables_search = [ [f'DA6_{group01}'], [f'DA6_{group02}'], ['type ipv6_addr'], ['flags dynamic,timeout'], ['chain VYOS_IPV6_INPUT_filter {'], ['type filter hook input priority filter', 'policy accept'], ['tcp dport 5151', f'update @DA6_{group01}', '{ ip6 saddr timeout 30s }', 'drop'], ['tcp dport 7272', f'ip6 saddr @DA6_{group01}', f'update @DA6_{group02}', '{ ip6 saddr timeout 5m }', 'drop'], ['tcp dport 22', f'ip6 saddr @DA6_{group02}', 'accept'] ] self.verify_nftables(nftables_search, 'ip6 vyos_filter') def test_ipv4_global_state(self): self.cli_set(['firewall', 'global-options', 'state-policy', 'established', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'related', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'invalid', 'action', 'drop']) self.cli_commit() nftables_search = [ ['jump VYOS_STATE_POLICY'], ['chain VYOS_STATE_POLICY'], ['ct state established', 'accept'], ['ct state invalid', 'drop'], ['ct state related', 'accept'] ] self.verify_nftables(nftables_search, 'ip vyos_filter') # Check conntrack is enabled from state-policy self.verify_nftables_chain([['accept']], 'ip vyos_conntrack', 'FW_CONNTRACK') self.verify_nftables_chain([['accept']], 'ip6 vyos_conntrack', 'FW_CONNTRACK') def test_ipv4_state_and_status_rules(self): name = 'smoketest-state' 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']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '1', 'state', 'related']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'action', 'reject']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '2', 'state', 'invalid']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'action', 'accept']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '3', 'state', 'new']) 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']) self.cli_set(['firewall', 'ipv4', 'name', name, 'rule', '4', 'state', 'established']) 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']) 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, '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', 'name', interface_in]) self.cli_set(['firewall', 'bridge', 'name', name, 'rule', '1', 'log']) 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', 'default-log']) 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 accept;'], [f'vlan id {vlan_id}', 'accept'], [f'vlan pcp {vlan_prior}', f'jump NAME_{name}'], ['log prefix "[bri-FWD-filter-default-D]"', 'drop', 'FWD-filter default-action drop'], [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'], ['accept', f'{name} default-action 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) ### Zone def test_zone_basic(self): self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'default-action', 'drop']) self.cli_set(['firewall', 'ipv6', 'name', 'smoketestv6', 'default-action', 'drop']) self.cli_set(['firewall', 'zone', 'smoketest-eth0', 'interface', 'eth0']) self.cli_set(['firewall', 'zone', 'smoketest-eth0', 'from', 'smoketest-local', 'firewall', 'name', 'smoketest']) self.cli_set(['firewall', 'zone', 'smoketest-eth0', 'intra-zone-filtering', 'firewall', 'ipv6-name', 'smoketestv6']) self.cli_set(['firewall', 'zone', 'smoketest-local', 'local-zone']) self.cli_set(['firewall', 'zone', 'smoketest-local', 'from', 'smoketest-eth0', 'firewall', 'name', 'smoketest']) self.cli_set(['firewall', 'global-options', 'state-policy', 'established', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'established', 'log']) self.cli_set(['firewall', 'global-options', 'state-policy', 'related', 'action', 'accept']) self.cli_set(['firewall', 'global-options', 'state-policy', 'invalid', 'action', 'drop']) self.cli_commit() nftables_search = [ ['chain VYOS_ZONE_FORWARD'], ['type filter hook forward priority filter + 1'], ['chain VYOS_ZONE_OUTPUT'], ['type filter hook output priority filter + 1'], ['chain VYOS_ZONE_LOCAL'], ['type filter hook input priority filter + 1'], ['chain VZONE_smoketest-eth0'], ['chain VZONE_smoketest-local_IN'], ['chain VZONE_smoketest-local_OUT'], ['oifname "eth0"', 'jump VZONE_smoketest-eth0'], ['jump VZONE_smoketest-local_IN'], ['jump VZONE_smoketest-local_OUT'], ['iifname "eth0"', 'jump NAME_smoketest'], ['oifname "eth0"', 'jump NAME_smoketest'], ['jump VYOS_STATE_POLICY'], ['chain VYOS_STATE_POLICY'], ['ct state established', 'log prefix "[STATE-POLICY-EST-A]"', 'accept'], ['ct state invalid', 'drop'], ['ct state related', 'accept'] ] nftables_search_v6 = [ ['chain VYOS_ZONE_FORWARD'], ['type filter hook forward priority filter + 1'], ['chain VYOS_ZONE_OUTPUT'], ['type filter hook output priority filter + 1'], ['chain VYOS_ZONE_LOCAL'], ['type filter hook input priority filter + 1'], ['chain VZONE_smoketest-eth0'], ['chain VZONE_smoketest-local_IN'], ['chain VZONE_smoketest-local_OUT'], ['oifname "eth0"', 'jump VZONE_smoketest-eth0'], ['jump VZONE_smoketest-local_IN'], ['jump VZONE_smoketest-local_OUT'], ['iifname "eth0"', 'jump NAME6_smoketestv6'], ['jump VYOS_STATE_POLICY6'], ['chain VYOS_STATE_POLICY6'], ['ct state established', 'log prefix "[STATE-POLICY-EST-A]"', 'accept'], ['ct state invalid', 'drop'], ['ct state related', 'accept'] ] self.verify_nftables(nftables_search, 'ip vyos_filter') self.verify_nftables(nftables_search_v6, 'ip6 vyos_filter') def test_flow_offload(self): self.cli_set(['interfaces', 'ethernet', 'eth0', 'vif', '10']) self.cli_set(['firewall', 'flowtable', 'smoketest', 'interface', 'eth0.10']) 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']) self.cli_set(['firewall', 'ipv4', 'forward', 'filter', 'rule', '1', 'state', 'related']) 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']) self.cli_set(['firewall', 'ipv6', 'forward', 'filter', 'rule', '1', 'state', 'related']) self.cli_commit() nftables_search = [ ['flowtable VYOS_FLOWTABLE_smoketest'], ['hook ingress priority filter'], ['devices = { eth0.10 }'], ['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') def test_zone_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', 'name', 'smoketest', 'rule', '1', 'action', 'offload']) self.cli_set(['firewall', 'ipv4', 'name', 'smoketest', 'rule', '1', 'offload-target', 'smoketest']) self.cli_set(['firewall', 'ipv6', 'name', 'smoketest', 'rule', '1', 'action', 'offload']) self.cli_set(['firewall', 'ipv6', 'name', 'smoketest', 'rule', '1', 'offload-target', 'smoketest']) self.cli_commit() nftables_search = [ ['chain NAME_smoketest'], ['flow add @VYOS_FLOWTABLE_smoketest'] ] self.verify_nftables(nftables_search, 'ip vyos_filter') nftables_search = [ ['chain NAME6_smoketest'], ['flow add @VYOS_FLOWTABLE_smoketest'] ] 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_system_ip.py b/smoketest/scripts/cli/test_system_ip.py index ac8b74236..5b0090237 100755 --- a/smoketest/scripts/cli/test_system_ip.py +++ b/smoketest/scripts/cli/test_system_ip.py @@ -1,137 +1,126 @@ #!/usr/bin/env python3 # # Copyright (C) 2020-2024 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.configsession import ConfigSessionError from vyos.utils.file import read_file base_path = ['system', 'ip'] class TestSystemIP(VyOSUnitTestSHIM.TestCase): def tearDown(self): self.cli_delete(base_path) self.cli_commit() def test_system_ip_forwarding(self): # Test if IPv4 forwarding can be disabled globally, default is '1' # which means forwarding enabled all_forwarding = '/proc/sys/net/ipv4/conf/all/forwarding' self.assertEqual(read_file(all_forwarding), '1') self.cli_set(base_path + ['disable-forwarding']) self.cli_commit() self.assertEqual(read_file(all_forwarding), '0') - def test_system_ip_directed_broadcast_forwarding(self): - # Test if IPv4 directed broadcast forwarding can be disabled globally, - # default is '1' which means forwarding enabled - bc_forwarding = '/proc/sys/net/ipv4/conf/all/bc_forwarding' - self.assertEqual(read_file(bc_forwarding), '1') - - self.cli_set(base_path + ['disable-directed-broadcast']) - self.cli_commit() - - self.assertEqual(read_file(bc_forwarding), '0') - def test_system_ip_multipath(self): # Test IPv4 multipathing options, options default to off -> '0' use_neigh = '/proc/sys/net/ipv4/fib_multipath_use_neigh' hash_policy = '/proc/sys/net/ipv4/fib_multipath_hash_policy' self.assertEqual(read_file(use_neigh), '0') self.assertEqual(read_file(hash_policy), '0') self.cli_set(base_path + ['multipath', 'ignore-unreachable-nexthops']) self.cli_set(base_path + ['multipath', 'layer4-hashing']) self.cli_commit() self.assertEqual(read_file(use_neigh), '1') self.assertEqual(read_file(hash_policy), '1') def test_system_ip_arp_table_size(self): # Maximum number of entries to keep in the ARP cache, the # default is 8k gc_thresh3 = '/proc/sys/net/ipv4/neigh/default/gc_thresh3' gc_thresh2 = '/proc/sys/net/ipv4/neigh/default/gc_thresh2' gc_thresh1 = '/proc/sys/net/ipv4/neigh/default/gc_thresh1' self.assertEqual(read_file(gc_thresh3), '8192') self.assertEqual(read_file(gc_thresh2), '4096') self.assertEqual(read_file(gc_thresh1), '1024') for size in [1024, 2048, 4096, 8192, 16384, 32768]: self.cli_set(base_path + ['arp', 'table-size', str(size)]) self.cli_commit() self.assertEqual(read_file(gc_thresh3), str(size)) self.assertEqual(read_file(gc_thresh2), str(size // 2)) self.assertEqual(read_file(gc_thresh1), str(size // 8)) def test_system_ip_protocol_route_map(self): protocols = ['any', 'babel', 'bgp', 'connected', 'eigrp', 'isis', 'kernel', 'ospf', 'rip', 'static', 'table'] for protocol in protocols: self.cli_set(['policy', 'route-map', f'route-map-{protocol}', 'rule', '10', 'action', 'permit']) self.cli_set(base_path + ['protocol', protocol, 'route-map', f'route-map-{protocol}']) self.cli_commit() # Verify route-map properly applied to FRR frrconfig = self.getFRRconfig('ip protocol', end='', daemon='zebra') for protocol in protocols: self.assertIn(f'ip protocol {protocol} route-map route-map-{protocol}', frrconfig) # Delete route-maps self.cli_delete(['policy', 'route-map']) self.cli_delete(base_path + ['protocol']) self.cli_commit() # Verify route-map properly applied to FRR frrconfig = self.getFRRconfig('ip protocol', end='', daemon='zebra') self.assertNotIn(f'ip protocol', frrconfig) def test_system_ip_protocol_non_existing_route_map(self): non_existing = 'non-existing' self.cli_set(base_path + ['protocol', 'static', 'route-map', non_existing]) # VRF does yet not exist - an error must be thrown with self.assertRaises(ConfigSessionError): self.cli_commit() self.cli_set(['policy', 'route-map', non_existing, 'rule', '10', 'action', 'deny']) # Commit again self.cli_commit() def test_system_ip_nht(self): self.cli_set(base_path + ['nht', 'no-resolve-via-default']) self.cli_commit() # Verify CLI config applied to FRR frrconfig = self.getFRRconfig('', end='', daemon='zebra') self.assertIn(f'no ip nht resolve-via-default', frrconfig) self.cli_delete(base_path + ['nht', 'no-resolve-via-default']) self.cli_commit() # Verify CLI config removed to FRR frrconfig = self.getFRRconfig('', end='', daemon='zebra') self.assertNotIn(f'no ip nht resolve-via-default', frrconfig) if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/src/conf_mode/firewall.py b/src/conf_mode/firewall.py index 3cf618363..e96e57154 100755 --- a/src/conf_mode/firewall.py +++ b/src/conf_mode/firewall.py @@ -1,523 +1,524 @@ #!/usr/bin/env python3 # # Copyright (C) 2021-2024 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 from glob import glob from sys import exit from vyos.base import Warning from vyos.config import Config from vyos.configdict import is_node_changed from vyos.configdiff import get_config_diff, Diff from vyos.configdep import set_dependents, call_dependents from vyos.configverify import verify_interface_exists from vyos.ethtool import Ethtool from vyos.firewall import fqdn_config_parse 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.process import call from vyos.utils.process import rc_cmd from vyos import ConfigError from vyos import airbag airbag.enable() nftables_conf = '/run/nftables.conf' sysfs_config = { 'all_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_all', 'enable': '0', 'disable': '1'}, 'broadcast_ping': {'sysfs': '/proc/sys/net/ipv4/icmp_echo_ignore_broadcasts', 'enable': '0', 'disable': '1'}, + 'directed_broadcast' : {'sysfs': '/proc/sys/net/ipv4/conf/all/bc_forwarding', 'enable': '1', 'disable': '0'}, 'ip_src_route': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_source_route'}, 'ipv6_receive_redirects': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_redirects'}, 'ipv6_src_route': {'sysfs': '/proc/sys/net/ipv6/conf/*/accept_source_route', 'enable': '0', 'disable': '-1'}, 'log_martians': {'sysfs': '/proc/sys/net/ipv4/conf/all/log_martians'}, 'receive_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/accept_redirects'}, 'send_redirects': {'sysfs': '/proc/sys/net/ipv4/conf/*/send_redirects'}, 'syn_cookies': {'sysfs': '/proc/sys/net/ipv4/tcp_syncookies'}, 'twa_hazards_protection': {'sysfs': '/proc/sys/net/ipv4/tcp_rfc1337'} } valid_groups = [ 'address_group', 'domain_group', 'network_group', 'port_group', 'interface_group' ] nested_group_types = [ 'address_group', 'network_group', 'mac_group', 'port_group', 'ipv6_address_group', 'ipv6_network_group' ] snmp_change_type = { 'unknown': 0, 'add': 1, 'delete': 2, 'change': 3 } snmp_event_source = 1 snmp_trap_mib = 'VYATTA-TRAP-MIB' snmp_trap_name = 'mgmtEventTrap' def geoip_updated(conf, firewall): diff = get_config_diff(conf) node_diff = diff.get_child_nodes_diff(['firewall'], expand_nodes=Diff.DELETE, recursive=True) out = { 'name': [], 'ipv6_name': [], 'deleted_name': [], 'deleted_ipv6_name': [] } updated = False for key, path in dict_search_recursive(firewall, 'geoip'): set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}' if (path[0] == 'ipv4'): out['name'].append(set_name) elif (path[0] == 'ipv6'): set_name = f'GEOIP_CC6_{path[1]}_{path[2]}_{path[4]}' out['ipv6_name'].append(set_name) updated = True if 'delete' in node_diff: for key, path in dict_search_recursive(node_diff['delete'], 'geoip'): set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}' if (path[0] == 'ipv4'): out['deleted_name'].append(set_name) elif (path[0] == 'ipv6'): set_name = f'GEOIP_CC_{path[1]}_{path[2]}_{path[4]}' out['deleted_ipv6_name'].append(set_name) updated = True if updated: return out return False def get_config(config=None): if config: conf = config else: conf = Config() base = ['firewall'] firewall = conf.get_config_dict(base, key_mangling=('-', '_'), no_tag_node_value_mangle=True, get_first_key=True, with_recursive_defaults=True) firewall['group_resync'] = bool('group' in firewall or is_node_changed(conf, base + ['group'])) if firewall['group_resync']: # Update nat and policy-route as firewall groups were updated set_dependents('group_resync', conf) firewall['geoip_updated'] = geoip_updated(conf, firewall) fqdn_config_parse(firewall) set_dependents('conntrack', conf) return firewall def verify_rule(firewall, rule_conf, ipv6): if 'action' not in rule_conf: raise ConfigError('Rule action must be defined') if 'jump' in rule_conf['action'] and 'jump_target' not in rule_conf: raise ConfigError('Action set to jump, but no jump-target specified') if 'jump_target' in rule_conf: if 'jump' not in rule_conf['action']: raise ConfigError('jump-target defined, but action jump needed and it is not defined') target = rule_conf['jump_target'] if not ipv6: if target not in dict_search_args(firewall, 'ipv4', 'name'): raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system') else: if target not in dict_search_args(firewall, 'ipv6', 'name'): raise ConfigError(f'Invalid jump-target. Firewall ipv6 name {target} does not exist on the system') if rule_conf['action'] == 'offload': if 'offload_target' not in rule_conf: raise ConfigError('Action set to offload, but no offload-target specified') offload_target = rule_conf['offload_target'] if not dict_search_args(firewall, 'flowtable', offload_target): raise ConfigError(f'Invalid offload-target. Flowtable "{offload_target}" does not exist on the system') if rule_conf['action'] != 'synproxy' and 'synproxy' in rule_conf: raise ConfigError('"synproxy" option allowed only for action synproxy') if rule_conf['action'] == 'synproxy': if 'state' in rule_conf: raise ConfigError('For action "synproxy" state cannot be defined') if not rule_conf.get('synproxy', {}).get('tcp'): raise ConfigError('synproxy TCP MSS is not defined') if rule_conf.get('protocol', {}) != 'tcp': raise ConfigError('For action "synproxy" the protocol must be set to TCP') if 'queue_options' in rule_conf: if 'queue' not in rule_conf['action']: raise ConfigError('queue-options defined, but action queue needed and it is not defined') if 'fanout' in rule_conf['queue_options'] and ('queue' not in rule_conf or '-' not in rule_conf['queue']): raise ConfigError('queue-options fanout defined, then queue needs to be defined as a range') if 'queue' in rule_conf and 'queue' not in rule_conf['action']: raise ConfigError('queue defined, but action queue needed and it is not defined') if 'fragment' in rule_conf: if {'match_frag', 'match_non_frag'} <= set(rule_conf['fragment']): raise ConfigError('Cannot specify both "match-frag" and "match-non-frag"') if 'limit' in rule_conf: if 'rate' in rule_conf['limit']: rate_int = re.sub(r'\D', '', rule_conf['limit']['rate']) if int(rate_int) < 1: raise ConfigError('Limit rate integer cannot be less than 1') if 'ipsec' in rule_conf: if {'match_ipsec', 'match_non_ipsec'} <= set(rule_conf['ipsec']): raise ConfigError('Cannot specify both "match-ipsec" and "match-non-ipsec"') if 'recent' in rule_conf: if not {'count', 'time'} <= set(rule_conf['recent']): raise ConfigError('Recent "count" and "time" values must be defined') tcp_flags = dict_search_args(rule_conf, 'tcp', 'flags') if tcp_flags: if dict_search_args(rule_conf, 'protocol') != 'tcp': raise ConfigError('Protocol must be tcp when specifying tcp flags') not_flags = dict_search_args(rule_conf, 'tcp', 'flags', 'not') if not_flags: duplicates = [flag for flag in tcp_flags if flag in not_flags] if duplicates: raise ConfigError(f'Cannot match a tcp flag as set and not set') if 'protocol' in rule_conf: if rule_conf['protocol'] == 'icmp' and ipv6: raise ConfigError(f'Cannot match IPv4 ICMP protocol on IPv6, use ipv6-icmp') if rule_conf['protocol'] == 'ipv6-icmp' and not ipv6: raise ConfigError(f'Cannot match IPv6 ICMP protocol on IPv4, use icmp') for side in ['destination', 'source']: if side in rule_conf: side_conf = rule_conf[side] if len({'address', 'fqdn', 'geoip'} & set(side_conf)) > 1: raise ConfigError('Only one of address, fqdn or geoip can be specified') if 'group' in side_conf: if len({'address_group', 'network_group', 'domain_group'} & set(side_conf['group'])) > 1: raise ConfigError('Only one address-group, network-group or domain-group can be specified') for group in valid_groups: if group in side_conf['group']: group_name = side_conf['group'][group] fw_group = f'ipv6_{group}' if ipv6 and group in ['address_group', 'network_group'] else group error_group = fw_group.replace("_", "-") if group in ['address_group', 'network_group', 'domain_group']: types = [t for t in ['address', 'fqdn', 'geoip'] if t in side_conf] if types: raise ConfigError(f'{error_group} and {types[0]} cannot both be defined') if group_name and group_name[0] == '!': group_name = group_name[1:] group_obj = dict_search_args(firewall, 'group', fw_group, group_name) if group_obj is None: raise ConfigError(f'Invalid {error_group} "{group_name}" on firewall rule') if not group_obj: Warning(f'{error_group} "{group_name}" has no members!') if 'port' in side_conf or dict_search_args(side_conf, 'group', 'port_group'): if 'protocol' not in rule_conf: raise ConfigError('Protocol must be defined if specifying a port or port-group') if rule_conf['protocol'] not in ['tcp', 'udp', 'tcp_udp']: raise ConfigError('Protocol must be tcp, udp, or tcp_udp when specifying a port or port-group') if 'port' in side_conf and dict_search_args(side_conf, 'group', 'port_group'): raise ConfigError(f'{side} port-group and port cannot both be defined') if 'add_address_to_group' in rule_conf: for type in ['destination_address', 'source_address']: if type in rule_conf['add_address_to_group']: if 'address_group' not in rule_conf['add_address_to_group'][type]: raise ConfigError(f'Dynamic address group must be defined.') else: target = rule_conf['add_address_to_group'][type]['address_group'] fwall_group = 'ipv6_address_group' if ipv6 else 'address_group' group_obj = dict_search_args(firewall, 'group', 'dynamic_group', fwall_group, target) if group_obj is None: raise ConfigError(f'Invalid dynamic address group on firewall rule') if 'log_options' in rule_conf: if 'log' not in rule_conf: raise ConfigError('log-options defined, but log is not enable') if 'snapshot_length' in rule_conf['log_options'] and 'group' not in rule_conf['log_options']: raise ConfigError('log-options snapshot-length defined, but log group is not define') if 'queue_threshold' in rule_conf['log_options'] and 'group' not in rule_conf['log_options']: raise ConfigError('log-options queue-threshold defined, but log group is not define') for direction in ['inbound_interface','outbound_interface']: if direction in rule_conf: if 'name' in rule_conf[direction] and 'group' in rule_conf[direction]: raise ConfigError(f'Cannot specify both interface group and interface name for {direction}') if 'group' in rule_conf[direction]: group_name = rule_conf[direction]['group'] if group_name[0] == '!': group_name = group_name[1:] group_obj = dict_search_args(firewall, 'group', 'interface_group', group_name) if group_obj is None: raise ConfigError(f'Invalid interface group "{group_name}" on firewall rule') if not group_obj: Warning(f'interface-group "{group_name}" has no members!') def verify_nested_group(group_name, group, groups, seen): if 'include' not in group: return seen.append(group_name) for g in group['include']: if g not in groups: raise ConfigError(f'Nested group "{g}" does not exist') if g in seen: raise ConfigError(f'Group "{group_name}" has a circular reference') if 'include' in groups[g]: verify_nested_group(g, groups[g], groups, seen) def verify_hardware_offload(ifname): ethtool = Ethtool(ifname) enabled, fixed = ethtool.get_hw_tc_offload() if not enabled and fixed: raise ConfigError(f'Interface "{ifname}" does not support hardware offload') if not enabled: raise ConfigError(f'Interface "{ifname}" requires "offload hw-tc-offload"') def verify(firewall): if 'flowtable' in firewall: for flowtable, flowtable_conf in firewall['flowtable'].items(): if 'interface' not in flowtable_conf: raise ConfigError(f'Flowtable "{flowtable}" requires at least one interface') for ifname in flowtable_conf['interface']: verify_interface_exists(ifname) if dict_search_args(flowtable_conf, 'offload') == 'hardware': interfaces = flowtable_conf['interface'] for ifname in interfaces: verify_hardware_offload(ifname) if 'group' in firewall: for group_type in nested_group_types: if group_type in firewall['group']: groups = firewall['group'][group_type] for group_name, group in groups.items(): verify_nested_group(group_name, group, groups, []) if 'ipv4' in firewall: for name in ['name','forward','input','output']: if name in firewall['ipv4']: for name_id, name_conf in firewall['ipv4'][name].items(): if 'jump' in name_conf['default_action'] and 'default_jump_target' not in name_conf: raise ConfigError('default-action set to jump, but no default-jump-target specified') if 'default_jump_target' in name_conf: target = name_conf['default_jump_target'] if 'jump' not in name_conf['default_action']: raise ConfigError('default-jump-target defined, but default-action jump needed and it is not defined') if name_conf['default_jump_target'] == name_id: raise ConfigError(f'Loop detected on default-jump-target.') ## Now need to check that default-jump-target exists (other firewall chain/name) if target not in dict_search_args(firewall['ipv4'], 'name'): raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system') if 'rule' in name_conf: for rule_id, rule_conf in name_conf['rule'].items(): verify_rule(firewall, rule_conf, False) if 'ipv6' in firewall: for name in ['name','forward','input','output']: if name in firewall['ipv6']: for name_id, name_conf in firewall['ipv6'][name].items(): if 'jump' in name_conf['default_action'] and 'default_jump_target' not in name_conf: raise ConfigError('default-action set to jump, but no default-jump-target specified') if 'default_jump_target' in name_conf: target = name_conf['default_jump_target'] if 'jump' not in name_conf['default_action']: raise ConfigError('default-jump-target defined, but default-action jump needed and it is not defined') if name_conf['default_jump_target'] == name_id: raise ConfigError(f'Loop detected on default-jump-target.') ## Now need to check that default-jump-target exists (other firewall chain/name) if target not in dict_search_args(firewall['ipv6'], 'name'): raise ConfigError(f'Invalid jump-target. Firewall name {target} does not exist on the system') if 'rule' in name_conf: for rule_id, rule_conf in name_conf['rule'].items(): verify_rule(firewall, rule_conf, True) #### ZONESSSS local_zone = False zone_interfaces = [] if 'zone' in firewall: for zone, zone_conf in firewall['zone'].items(): if 'local_zone' not in zone_conf and 'interface' not in zone_conf: raise ConfigError(f'Zone "{zone}" has no interfaces and is not the local zone') if 'local_zone' in zone_conf: if local_zone: raise ConfigError('There cannot be multiple local zones') if 'interface' in zone_conf: raise ConfigError('Local zone cannot have interfaces assigned') if 'intra_zone_filtering' in zone_conf: raise ConfigError('Local zone cannot use intra-zone-filtering') local_zone = True if 'interface' in zone_conf: found_duplicates = [intf for intf in zone_conf['interface'] if intf in zone_interfaces] if found_duplicates: raise ConfigError(f'Interfaces cannot be assigned to multiple zones') zone_interfaces += zone_conf['interface'] if 'intra_zone_filtering' in zone_conf: intra_zone = zone_conf['intra_zone_filtering'] if len(intra_zone) > 1: raise ConfigError('Only one intra-zone-filtering action must be specified') if 'firewall' in intra_zone: v4_name = dict_search_args(intra_zone, 'firewall', 'name') if v4_name and not dict_search_args(firewall, 'ipv4', 'name', v4_name): raise ConfigError(f'Firewall name "{v4_name}" does not exist') v6_name = dict_search_args(intra_zone, 'firewall', 'ipv6_name') if v6_name and not dict_search_args(firewall, 'ipv6', 'name', v6_name): raise ConfigError(f'Firewall ipv6-name "{v6_name}" does not exist') if not v4_name and not v6_name: raise ConfigError('No firewall names specified for intra-zone-filtering') if 'from' in zone_conf: for from_zone, from_conf in zone_conf['from'].items(): if from_zone not in firewall['zone']: raise ConfigError(f'Zone "{zone}" refers to a non-existent or deleted zone "{from_zone}"') v4_name = dict_search_args(from_conf, 'firewall', 'name') if v4_name and not dict_search_args(firewall, 'ipv4', 'name', v4_name): raise ConfigError(f'Firewall name "{v4_name}" does not exist') v6_name = dict_search_args(from_conf, 'firewall', 'ipv6_name') if v6_name and not dict_search_args(firewall, 'ipv6', 'name', v6_name): raise ConfigError(f'Firewall ipv6-name "{v6_name}" does not exist') return None def generate(firewall): if not os.path.exists(nftables_conf): firewall['first_install'] = True if 'zone' in firewall: for local_zone, local_zone_conf in firewall['zone'].items(): if 'local_zone' not in local_zone_conf: continue local_zone_conf['from_local'] = {} for zone, zone_conf in firewall['zone'].items(): if zone == local_zone or 'from' not in zone_conf: continue if local_zone in zone_conf['from']: local_zone_conf['from_local'][zone] = zone_conf['from'][local_zone] render(nftables_conf, 'firewall/nftables.j2', firewall) return None def apply_sysfs(firewall): for name, conf in sysfs_config.items(): paths = glob(conf['sysfs']) value = None if name in firewall['global_options']: conf_value = firewall['global_options'][name] if conf_value in conf: value = conf[conf_value] elif conf_value == 'enable': value = '1' elif conf_value == 'disable': value = '0' if value: for path in paths: with open(path, 'w') as f: f.write(value) def apply(firewall): install_result, output = rc_cmd(f'nft --file {nftables_conf}') if install_result == 1: raise ConfigError(f'Failed to apply firewall: {output}') apply_sysfs(firewall) call_dependents() # T970 Enable a resolver (systemd daemon) that checks # domain-group/fqdn addresses and update entries for domains by timeout # If router loaded without internet connection or for synchronization domain_action = 'stop' if dict_search_args(firewall, 'group', 'domain_group') or firewall['ip_fqdn'] or firewall['ip6_fqdn']: domain_action = 'restart' call(f'systemctl {domain_action} vyos-domain-resolver.service') if firewall['geoip_updated']: # Call helper script to Update set contents if 'name' in firewall['geoip_updated'] or 'ipv6_name' in firewall['geoip_updated']: print('Updating GeoIP. Please wait...') geoip_update(firewall) return None if __name__ == '__main__': try: c = get_config() verify(c) generate(c) apply(c) except ConfigError as e: print(e) exit(1) diff --git a/src/conf_mode/system_ip.py b/src/conf_mode/system_ip.py index b945b51f2..2a0bda91a 100755 --- a/src/conf_mode/system_ip.py +++ b/src/conf_mode/system_ip.py @@ -1,143 +1,138 @@ #!/usr/bin/env python3 # # Copyright (C) 2019-2024 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/>. from sys import exit from vyos.config import Config from vyos.configdict import dict_merge from vyos.configverify import verify_route_map from vyos.template import render_to_string from vyos.utils.dict import dict_search from vyos.utils.file import write_file from vyos.utils.process import is_systemd_service_active from vyos.utils.system import sysctl_write from vyos import ConfigError from vyos import frr from vyos import airbag airbag.enable() def get_config(config=None): if config: conf = config else: conf = Config() base = ['system', 'ip'] opt = conf.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True, with_recursive_defaults=True) # When working with FRR we need to know the corresponding address-family opt['afi'] = 'ip' # We also need the route-map information from the config # # XXX: one MUST always call this without the key_mangling() option! See # vyos.configverify.verify_common_route_maps() for more information. tmp = {'policy' : {'route-map' : conf.get_config_dict(['policy', 'route-map'], get_first_key=True)}} # Merge policy dict into "regular" config dict opt = dict_merge(tmp, opt) return opt def verify(opt): if 'protocol' in opt: for protocol, protocol_options in opt['protocol'].items(): if 'route_map' in protocol_options: verify_route_map(protocol_options['route_map'], opt) return def generate(opt): opt['frr_zebra_config'] = render_to_string('frr/zebra.route-map.frr.j2', opt) return def apply(opt): # Apply ARP threshold values # table_size has a default value - thus the key always exists size = int(dict_search('arp.table_size', opt)) # Amount upon reaching which the records begin to be cleared immediately sysctl_write('net.ipv4.neigh.default.gc_thresh3', size) # Amount after which the records begin to be cleaned after 5 seconds sysctl_write('net.ipv4.neigh.default.gc_thresh2', size // 2) # Minimum number of stored records is indicated which is not cleared sysctl_write('net.ipv4.neigh.default.gc_thresh1', size // 8) # enable/disable IPv4 forwarding tmp = dict_search('disable_forwarding', opt) value = '0' if (tmp != None) else '1' write_file('/proc/sys/net/ipv4/conf/all/forwarding', value) - # enable/disable IPv4 directed broadcast forwarding - tmp = dict_search('disable_directed_broadcast', opt) - value = '0' if (tmp != None) else '1' - write_file('/proc/sys/net/ipv4/conf/all/bc_forwarding', value) - # configure multipath tmp = dict_search('multipath.ignore_unreachable_nexthops', opt) value = '1' if (tmp != None) else '0' sysctl_write('net.ipv4.fib_multipath_use_neigh', value) tmp = dict_search('multipath.layer4_hashing', opt) value = '1' if (tmp != None) else '0' sysctl_write('net.ipv4.fib_multipath_hash_policy', value) # configure TCP options (defaults as of Linux 6.4) tmp = dict_search('tcp.mss.probing', opt) if tmp is None: value = 0 elif tmp == 'on-icmp-black-hole': value = 1 elif tmp == 'force': value = 2 else: # Shouldn't happen raise ValueError("TCP MSS probing is neither 'on-icmp-black-hole' nor 'force'!") sysctl_write('net.ipv4.tcp_mtu_probing', value) tmp = dict_search('tcp.mss.base', opt) value = '1024' if (tmp is None) else tmp sysctl_write('net.ipv4.tcp_base_mss', value) tmp = dict_search('tcp.mss.floor', opt) value = '48' if (tmp is None) else tmp sysctl_write('net.ipv4.tcp_mtu_probe_floor', value) # During startup of vyos-router that brings up FRR, the service is not yet # running when this script is called first. Skip this part and wait for initial # commit of the configuration to trigger this statement if is_systemd_service_active('frr.service'): zebra_daemon = 'zebra' # Save original configuration prior to starting any commit actions frr_cfg = frr.FRRConfig() # The route-map used for the FIB (zebra) is part of the zebra daemon frr_cfg.load_configuration(zebra_daemon) frr_cfg.modify_section(r'no ip nht resolve-via-default') frr_cfg.modify_section(r'ip protocol \w+ route-map [-a-zA-Z0-9.]+', stop_pattern='(\s|!)') if 'frr_zebra_config' in opt: frr_cfg.add_before(frr.default_add_before, opt['frr_zebra_config']) frr_cfg.commit_configuration(zebra_daemon) if __name__ == '__main__': try: c = get_config() verify(c) generate(c) apply(c) except ConfigError as e: print(e) exit(1) diff --git a/src/migration-scripts/firewall/14-to-15 b/src/migration-scripts/firewall/14-to-15 new file mode 100755 index 000000000..735839365 --- /dev/null +++ b/src/migration-scripts/firewall/14-to-15 @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2022-2024 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/>. + +# T5535: Migrate <set system ip disable-directed-broadcast> to <set firewall global-options directed-broadcas [enable|disable] + +from sys import argv +from sys import exit + +from vyos.configtree import ConfigTree + +if len(argv) < 2: + print("Must specify file name!") + exit(1) + +file_name = argv[1] + +with open(file_name, 'r') as f: + config_file = f.read() + +config = ConfigTree(config_file) + +base = ['firewall'] + +if config.exists(['system', 'ip', 'disable-directed-broadcast']): + config.set(['firewall', 'global-options', 'directed-broadcast'], value='disable') + config.delete(['system', 'ip', 'disable-directed-broadcast']) + +try: + with open(file_name, 'w') as f: + f.write(config.to_string()) +except OSError as e: + print("Failed to save the modified config: {}".format(e)) + exit(1) \ No newline at end of file