From bceeaa0c1f038b7b69707c03a3bdbbab86674e66 Mon Sep 17 00:00:00 2001 From: Brad House Date: Thu, 8 Oct 2026 12:42:55 +0000 Subject: [PATCH 1/2] Match all ports for a tcp or udp ACL rule given without ports A VPC network ACL rule for tcp or udp without a port range reaches the VR as the range 0:0: TcpAclRule and UdpAclRule hold their ports as ints, so they are always sent. AclRule.init_vpc() only checked that first_port was present and rendered "-m tcp --dport 0", so the rule matched port 0 only: an allow admitted nothing, and a deny blocked nothing, letting the traffic through to the rules after it. The nftables paths (routed IPv4, IPv6) already treat 0 as no ports and match all of them. Treat 0:0 as no ports on the iptables path too. A range starting at 0, such as 0:1024, is still applied as given. Fixes: #14363 Signed-off-by: Brad House --- systemvm/debian/opt/cloud/bin/configure.py | 4 +- systemvm/test/TestCsAclPorts.py | 82 ++++++++++++++++++++++ 2 files changed, 85 insertions(+), 1 deletion(-) create mode 100644 systemvm/test/TestCsAclPorts.py diff --git a/systemvm/debian/opt/cloud/bin/configure.py b/systemvm/debian/opt/cloud/bin/configure.py index bf48be66694c..ffc98923fd0a 100755 --- a/systemvm/debian/opt/cloud/bin/configure.py +++ b/systemvm/debian/opt/cloud/bin/configure.py @@ -632,7 +632,9 @@ def init_vpc(self, direction, acl, rule, config): self.dport = "" if 'allowed' in list(rule.keys()) and rule['allowed']: self.action = "ACCEPT" - if 'first_port' in list(rule.keys()): + # A tcp or udp rule without ports is sent with the range 0:0; like the nft paths, match all ports + no_ports = rule.get('first_port', 0) == 0 and rule.get('last_port', 0) == 0 + if 'first_port' in list(rule.keys()) and not no_ports: self.dport = "-m %s --dport %s" % (self.protocol, rule['first_port']) if 'last_port' in list(rule.keys()) and self.dport and \ rule['last_port'] != rule['first_port']: diff --git a/systemvm/test/TestCsAclPorts.py b/systemvm/test/TestCsAclPorts.py new file mode 100644 index 000000000000..3fd3eed17912 --- /dev/null +++ b/systemvm/test/TestCsAclPorts.py @@ -0,0 +1,82 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import unittest +from configure import CsAcl + + +class FakeConfig: + + def __init__(self): + self.fw = [] + self.ipv6_acl = [] + self.nft_ipv4_acl = [] + + def get_fw(self): + return self.fw + + def get_ipv6_acl(self): + return self.ipv6_acl + + def get_nft_ipv4_acl(self): + return self.nft_ipv4_acl + + def is_vpc(self): + return True + + def is_routed(self): + return False + + def get_ingress_chain(self, device, ip): + return "ACL_INBOUND_%s" % device + + def get_egress_chain(self, device, ip): + return "ACL_OUTBOUND_%s" % device + + def get_egress_table(self): + return "mangle" + + +class TestCsAclPorts(unittest.TestCase): + + def iptables_rules(self, ingress, egress=None): + config = FakeConfig() + obj = {"device": "eth3", "nic_ip": "10.1.1.1", "nic_netmask": "24", + "ingress_rules": ingress, "egress_rules": egress or []} + CsAcl.AclDevice(obj, config).create() + return [fw[2] for fw in config.fw] + + def test_tcp_and_udp_without_ports_match_all_ports(self): + # a tcp or udp rule without ports reaches the VR as the port range 0:0 + rules = self.iptables_rules( + [{"type": "tcp", "cidr": "10.0.0.0/8", "first_port": 0, "last_port": 0, "allowed": True}], + [{"type": "udp", "cidr": "10.0.0.0/8", "first_port": 0, "last_port": 0, "allowed": False}]) + self.assertEqual(rules, ["-A ACL_INBOUND_eth3 -p tcp -s 10.0.0.0/8 -j ACCEPT", + "-A ACL_OUTBOUND_eth3 -p udp -d 10.0.0.0/8 -j DROP"]) + + def test_ports_and_port_ranges_are_kept(self): + rules = self.iptables_rules([ + {"type": "tcp", "cidr": "10.0.0.0/8", "first_port": 22, "last_port": 22, "allowed": True}, + {"type": "udp", "cidr": "10.0.0.0/8", "first_port": 1000, "last_port": 2000, "allowed": True}, + {"type": "tcp", "cidr": "10.0.0.0/8", "first_port": 0, "last_port": 1024, "allowed": False}]) + self.assertEqual(rules, ["-A ACL_INBOUND_eth3 -p tcp -s 10.0.0.0/8 -m tcp --dport 22 -j ACCEPT", + "-A ACL_INBOUND_eth3 -p udp -s 10.0.0.0/8 -m udp --dport 1000:2000 -j ACCEPT", + "-A ACL_INBOUND_eth3 -p tcp -s 10.0.0.0/8 -m tcp --dport 0:1024 -j DROP"]) + + +if __name__ == '__main__': + unittest.main() From c14a4518f14386903e2abca279ef18f56accb345 Mon Sep 17 00:00:00 2001 From: Brad House Date: Thu, 8 Oct 2026 12:52:36 +0000 Subject: [PATCH 2/2] Test the IPv6 rendering and other rule types alongside the port fix Signed-off-by: Brad House --- systemvm/test/TestCsAclPorts.py | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/systemvm/test/TestCsAclPorts.py b/systemvm/test/TestCsAclPorts.py index 3fd3eed17912..3ae9f4757b4c 100644 --- a/systemvm/test/TestCsAclPorts.py +++ b/systemvm/test/TestCsAclPorts.py @@ -53,12 +53,15 @@ def get_egress_table(self): class TestCsAclPorts(unittest.TestCase): - def iptables_rules(self, ingress, egress=None): + def acl(self, ingress, egress=None): config = FakeConfig() - obj = {"device": "eth3", "nic_ip": "10.1.1.1", "nic_netmask": "24", + obj = {"device": "eth3", "nic_ip": "10.1.1.1", "nic_netmask": "24", "nic_ip6_cidr": "fd00:1::/64", "ingress_rules": ingress, "egress_rules": egress or []} CsAcl.AclDevice(obj, config).create() - return [fw[2] for fw in config.fw] + return config + + def iptables_rules(self, ingress, egress=None): + return [fw[2] for fw in self.acl(ingress, egress).fw] def test_tcp_and_udp_without_ports_match_all_ports(self): # a tcp or udp rule without ports reaches the VR as the port range 0:0 @@ -68,6 +71,21 @@ def test_tcp_and_udp_without_ports_match_all_ports(self): self.assertEqual(rules, ["-A ACL_INBOUND_eth3 -p tcp -s 10.0.0.0/8 -j ACCEPT", "-A ACL_OUTBOUND_eth3 -p udp -d 10.0.0.0/8 -j DROP"]) + def test_tcp_without_ports_matches_all_ports_on_ipv6_too(self): + config = self.acl([{"type": "tcp", "cidr": "10.0.0.0/8,fd00:2::/64", "first_port": 0, "last_port": 0, "allowed": True}]) + self.assertEqual([fw[2] for fw in config.fw], ["-A ACL_INBOUND_eth3 -p tcp -s 10.0.0.0/8 -j ACCEPT"]) + ipv6 = [r['rule'] for r in config.ipv6_acl if r.get('chain') == "eth3_ingress_policy" and 'rule' in r] + self.assertIn("ip6 saddr fd00:2::/64 tcp dport { 0-65535 } accept", ipv6) + + def test_rules_without_port_fields_are_unaffected(self): + rules = self.iptables_rules([ + {"type": "icmp", "cidr": "10.0.0.0/8", "icmp_type": 8, "icmp_code": 0, "allowed": True}, + {"type": "protocol", "protocol": 47, "cidr": "10.0.0.0/8", "allowed": True}, + {"type": "all", "cidr": "10.0.0.0/8", "allowed": False}]) + self.assertEqual(rules, ["-A ACL_INBOUND_eth3 -p icmp -s 10.0.0.0/8 -m icmp --icmp-type 8/0 -j ACCEPT", + "-A ACL_INBOUND_eth3 -p 47 -s 10.0.0.0/8 -j ACCEPT", + "-A ACL_INBOUND_eth3 -p all -s 10.0.0.0/8 -j DROP"]) + def test_ports_and_port_ranges_are_kept(self): rules = self.iptables_rules([ {"type": "tcp", "cidr": "10.0.0.0/8", "first_port": 22, "last_port": 22, "allowed": True},