Merge pull request #2924 from vyos/mergify/bp/sagitta/pr-2756

T4839: firewall: Add dynamic address group in firewall configuration (backport #2756)

1 files changed, 81 insertions, 0 deletions

diff --git a/smoketest/scripts/cli/test_firewall.py b/smoketest/scripts/cli/test_firewall.py
index 72fbdb37d..a7dd11145 100755
--- a/smoketest/scripts/cli/test_firewall.py
+++ b/smoketest/scripts/cli/test_firewall.py
@@ -403,6 +403,46 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase):
 
         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'
@@ -540,6 +580,47 @@ class TestFirewall(VyOSUnitTestSHIM.TestCase):
 
         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_state_and_status_rules(self):
         name = 'smoketest-state'
         interface = 'eth0'