Match IPv6 extension header protocol numbers as extension headers

An ACL rule given by the protocol number of an IPv6 extension header
(0, 43, 44, 60, 135) is rendered as nft's "exthdr <name> exists" for
that header, rather than as an upper-layer protocol.

Signed-off-by: Brad House <bhouse@nexthop.ai>

Author: bhouse-nexthop

systemvm/debian/opt/cloud/bin/configure.py

@@ -71,6 +71,17 @@ def removeUndesiredCidrs(cidrs, version):
     return None
 
 
+# nft names of the IPv6 extension headers, which can also be given as an ACL protocol number
+IP6_EXTHDR_NAMES = {0: "hbh", 43: "rt", 44: "frag", 60: "dst", 135: "mh"}
+
+
+def ip6ProtocolMatch(number):
+    """ An extension header number is matched as that extension header, any other as the protocol """
+    if number in IP6_EXTHDR_NAMES:
+        return "exthdr %s exists" % IP6_EXTHDR_NAMES[number]
+    return "meta l4proto %d" % number
+
+
 def appendStringIfNotEmpty(s1, s2):
     if s2:
         if not isinstance(s2, str):
@@ -527,7 +538,7 @@ def __process_ip6(self, direction, rule_list):
                 if protocol != "all":
                     icmp_type = ""
                     if protocol == "protocol":
-                        protocol = "meta l4proto %d" % rule['protocol']
+                        protocol = ip6ProtocolMatch(rule['protocol'])
                     proto = protocol
                     if proto == "icmp":
                         proto = proto_str = "icmpv6"

systemvm/test/TestCsAclIpv6.py

@@ -58,6 +58,18 @@ def test_protocol_number_rule_uses_meta_l4proto(self):
         egress = [r['rule'] for r in config.ipv6_acl if r.get('chain') == "eth3_egress_policy" and 'rule' in r]
         self.assertIn("ip6 daddr 2001:db8:1::/64 meta l4proto 50 accept", egress)
 
+    def test_extension_header_number_rule(self):
+        config = FakeConfig()
+        rules = [{"type": "protocol", "protocol": p, "cidr": "2001:db8::/64", "allowed": False} for p in (0, 43, 44, 60, 135)]
+        obj = {"device": "eth3", "nic_ip": "10.1.1.1", "nic_netmask": "24", "nic_ip6_cidr": "fd00:1::/64",
+               "ingress_rules": rules, "egress_rules": []}
+        acl = CsAcl.AclDevice(obj, config)
+        acl.create()
+
+        ingress = [r['rule'] for r in config.ipv6_acl if r.get('chain') == "eth3_ingress_policy" and 'rule' in r]
+        for name in ("hbh", "rt", "frag", "dst", "mh"):
+            self.assertIn("ip6 saddr 2001:db8::/64 exthdr %s exists drop" % name, ingress)
+
 
 if __name__ == '__main__':
     unittest.main()