Nftables ensure no duplicate rules inside an individual chain can exist. Skip IPv4 ICMP loop on mixed family on ICMPv6 only terms.

PiperOrigin-RevId: 504633020

Author: TheLinuxGuy

capirca/lib/nftables.py

@@ -133,6 +133,7 @@ def MapICMPtypes(self, af, term_icmp_types):
     Function is used inside PortsAndProtocols.
 
     Args:
+      af: address family.
       term_icmp_types: ICMP types keywords.
 
     Returns:
@@ -462,6 +463,7 @@ def RulesetGenerator(self, term):
       list of strings. Representing a ruleset for later formatting.
     """
     term_ruleset = []
+    unique_term_ruleset = []
     comment = 'comment '
 
     # COMMENT handling.
@@ -488,11 +490,19 @@ def RulesetGenerator(self, term):
                                                self.term.destination_port,
                                                self.term.icmp_type)
 
+      # Do not render ICMP types if IP family mismatch.
+      if ((address_family == 'ip6' and 'icmp' in self.term.protocol) or
+          (address_family == 'ip' and ('icmpv6' in self.term.protocol)
+           or 'icmp6' in self.term.protocol)):
+        continue
       # TODO: If verdict is not supported, drop nftable_rule for it.
       nftable_rule = self.GroupExpressions(address_list, proto_and_ports, opt,
                                            verdict, comment)
       term_ruleset.extend(nftable_rule)
-    return term_ruleset
+    # Ensure that chain statements contain no duplicates rules.
+    unique_term_ruleset = [
+        i for n, i in enumerate(term_ruleset) if i not in term_ruleset[:n]]
+    return unique_term_ruleset
 
   def _AddressClassifier(self, address_to_classify):
     """Organizes network addresses according to IP family in a dict.
@@ -771,7 +781,7 @@ def _ConfigurationDictionary(self, nft_pol):
     for (header, base_chain_name, nf_af, nf_hook, nf_priority,
          filter_policy_default_action, verbose, child_chains) in nft_pol:
       base_chain_comment = ''
-      # Add max character checking on header.comment later if needed.
+      # TODO: If child_chain ruleset is empty don't store term.
       if verbose:
         base_chain_comment = header.comment
       nftables[nf_af][base_chain_name] = {

tests/lib/nftables_test.py

@@ -14,6 +14,7 @@
 """Unittest for Nftables rendering module."""
 
 import datetime
+import re
 from unittest import mock
 from absl import logging
 from absl.testing import absltest
@@ -129,7 +130,24 @@ def __init__(self, in_dict: dict):
 }
 """
 
-# TODO(gfm): Noverbose testing once Term handling is added.
+HEADER_MIXED_AF = """
+header {
+  target:: nftables mixed output
+}
+"""
+
+HEADER_IPV4_AF = """
+header {
+  target:: nftables inet output
+}
+"""
+
+HEADER_IPV6_AF = """
+header {
+  target:: nftables inet6 output
+}
+"""
+
 HEADER_NOVERBOSE = """
 header {
   target:: nftables mixed output noverbose
@@ -154,6 +172,13 @@ def __init__(self, in_dict: dict):
 }
 """
 
+DENY_TERM = """
+term deny-term {
+  comment:: "Dual-stack IPv4/v6 deny all"
+  action:: deny
+}
+"""
+
 ESTABLISHED_OPTION_TERM = """
 term established-term {
   protocol:: udp
@@ -541,6 +566,36 @@ def testSkippedTerm(self, termdata, messagetxt):
     record = ctx.records[1]
     self.assertEqual(record.message, messagetxt)
 
+  @parameterized.parameters(
+      (HEADER_MIXED_AF + ICMPV6_TERM, 'ip protocol icmp'),
+      (HEADER_IPV4_AF + ICMPV6_TERM, 'meta l4proto icmpv6'),
+      (HEADER_IPV6_AF + ICMP_TERM, 'ip protocol icmp'),
+  )
+  def testRulesetGeneratorICMPmismatch(self, pol_data, doesnotcontain):
+    # This test ensures that ICMPv6 only term isn't rendered in a mixed header.
+    nftables.Nftables(
+        policy.ParsePolicy(pol_data, self.naming), EXP_INFO)
+    nft = str(
+        nftables.Nftables(
+            policy.ParsePolicy(pol_data, self.naming), EXP_INFO))
+    self.assertNotRegex(nft, doesnotcontain)
+
+  def testRulesetGeneratorUniqueChain(self):
+    # This test is intended to verify that on mixed address family rulesets
+    # no duplicate instance of a simple deny is rendered within a mixed chain.
+    expected_term_rule = 'drop comment "Dual-stack IPv4/v6 deny all"'
+    count = 0
+    nftables.Nftables(
+        policy.ParsePolicy(HEADER_MIXED_AF + DENY_TERM, self.naming), EXP_INFO)
+    nft = str(
+        nftables.Nftables(
+            policy.ParsePolicy(
+                HEADER_MIXED_AF + DENY_TERM, self.naming), EXP_INFO))
+    matching_lines = re.findall(expected_term_rule, nft)
+    for match in matching_lines:
+      count += 1
+    self.assertEqual(count, 1)
+
   @parameterized.parameters(
       (GOOD_HEADER_1 + GOOD_TERM_2, 'inet6'),
       (GOOD_HEADER_1 + ICMPV6_TERM, 'inet6'),