Nftables implement support for source-interface and destination-interface. Add tests to counter and logging tokens.

PiperOrigin-RevId: 505237024

Author: TheLinuxGuy

capirca/lib/nftables.py

@@ -1,4 +1,4 @@
-# Copyright 2022 Google Inc. All Rights Reserved.
+# Copyright 2023 Google Inc. All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -350,14 +350,17 @@ def _OptionsHandler(self, term):
     else:
       return ''
 
-  def GroupExpressions(self, address_expr, pp_expr, options, verdict, comment):
+  def GroupExpressions(
+      self, int_expr, address_expr, pp_expr, options, verdict, comment
+  ):
     """Combines all expressions with a verdict (decision).
 
     The inputs are already pre-sanitized by RulesetGenerator. NFTables processes
     rules from left-to-right - ending in a verdict. We form our ruleset then
     towards the end append any term.options from _OptionsHandler.
 
     Args:
+      int_expr: RulesetGenerator source or destination interface str.
       address_expr: pre-processed list of nftable statements of network
         addresses.
       pp_expr: pre-processed list of nftables protocols and ports.
@@ -369,7 +372,6 @@ def GroupExpressions(self, address_expr, pp_expr, options, verdict, comment):
       list of strings representing valid nftables statements.
     """
     statement = []
-    statement_with_comment = []
     if address_expr:
       for addr in address_expr:
         if pp_expr:
@@ -392,12 +394,14 @@ def GroupExpressions(self, address_expr, pp_expr, options, verdict, comment):
     else:
       # If no addresses or ports & protocol. Verdict only statement.
       statement.append((Add(options) + Add(verdict)))
+    # source/destination interface handling always to be done at the end.
+    if int_expr:
+      # 'statement' is a list because join to another list in RulesetGenerator.
+      statement[0] = int_expr + Add(statement[0])
     # Handling of comments should always be done after verdict statement.
-    if comment != 'comment ':
-      statement_with_comment.append(statement[0] + Add(comment))
-      return statement_with_comment
-    else:
-      return statement
+    if comment:
+      statement[0] = statement[0] + Add(comment)
+    return statement
 
   def _AddrStatement(self, address_family, src_addr, dst_addr):
     """Builds an NFTables address statement.
@@ -464,12 +468,20 @@ def RulesetGenerator(self, term):
     """
     term_ruleset = []
     unique_term_ruleset = []
-    comment = 'comment '
+    comment = ''
 
     # COMMENT handling.
     if self.verbose:
-      comment += aclgenerator.TruncateWords(
+      comment = 'comment ' + aclgenerator.TruncateWords(
           self.term.comment, Nftables.COMMENT_CHAR_LIMIT)
+
+    # INTERFACE (source/destination) handling.
+    if term.source_interface:
+      interface = 'iifname' + Add(term.source_interface)
+    elif term.destination_interface:
+      interface = 'oifname' + Add(term.destination_interface)
+    else:
+      interface = ''
     # OPTIONS / LOGGING / COUNTERS
     opt = self._OptionsHandler(term)
     # STATEMENT VERDICT / ACTION.
@@ -482,22 +494,28 @@ def RulesetGenerator(self, term):
       address_list = self._AddrStatement(address_family,
                                          self.term.source_address,
                                          self.term.destination_address)
+      # Check if we're dealing with a term of a different IP family that needs
+      # to be skipped.
+      if not address_list and (
+          self.term.source_address or self.term.destination_address):
+        continue
 
       # PORTS and PROTOCOLS handling.
       proto_and_ports = self.PortsAndProtocols(address_family,
                                                self.term.protocol,
                                                self.term.source_port,
                                                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)
+      nftable_rule = self.GroupExpressions(
+          interface, address_list, proto_and_ports, opt, verdict, comment
+      )
       term_ruleset.extend(nftable_rule)
     # Ensure that chain statements contain no duplicates rules.
     unique_term_ruleset = [
@@ -610,9 +628,11 @@ def _BuildTokens(self):
         'protocol',
         'platform',
         'platform_exclude',
+        'source_interface', # NFT iifname
         'source_address',
         'source_address_exclude',
         'source_port',
+        'destination_interface', # NFT oifname
         'translated',  # obj attribute, not token
         'stateless_reply',
     }
@@ -666,11 +686,16 @@ def _TranslatePolicy(self, pol, exp_info):
       child_chains = collections.defaultdict(dict)
       term_names = set()
       new_terms = []
-      # TODO: Add checks for ICMP and address families.
       for term in terms:
         if term.name in term_names:
           raise TermError('Duplicate term name')
         term_names.add(term.name)
+        if term.source_interface and term.destination_interface:
+          raise TermError(
+              'Incorrect interface on term. Must be either be a source or'
+              ' destination, not both.'
+          )
+          continue
         if term.stateless_reply:
           logging.warning(
               'WARNING: Term %s is a stateless reply '

doc/generators/nftables.md

@@ -37,12 +37,16 @@ When reporting bugs about this generator ensure to include:
 - _destination-port::_ One or more service definition tokens.
 - _expiration::_ stop rendering this term after specified date. [YYYY](YYYY.md)-[MM](MM.md)-[DD](DD.md)
 - _icmp-type::_ Specify icmp-type code to match.
+- _source-interface::_ input direction interface name (renders as: [iifname](https://wiki.nftables.org/wiki-nftables/index.php/Matching_packet_metainformation))
 - _source-address::_ One or more source address tokens.
 - _source-port::_ One or more service definition tokens.
+- _destination-interface::_ output direction interface name (renders as: [oifname](https://wiki.nftables.org/wiki-nftables/index.php/Matching_packet_metainformation))
 - _protocol::_ The network protocol(s) this term will match.
 - _logging::_ NFTables system logging (host-based).
 - _counter::_ NFTables counter for specific term.
 
+Note: combining source-interface and destination-interface tokens within a term is not supported.
+
 ## Sub-tokens
 
 ### Actions

tests/lib/nftables_test.py

@@ -55,9 +55,11 @@ def __init__(self, in_dict: dict):
     'protocol',
     'platform',
     'platform_exclude',
+    'source_interface', #input interface
     'source_address',
     'source_address_exclude',
     'source_port',
+    'destination_interface', #ouput interface
     'translated',  # obj attribute, not token
     'stateless_reply',
 })
@@ -110,6 +112,10 @@ def __init__(self, in_dict: dict):
         'version-2-multicast-listener-report',
     },
 }
+# IP address data, to be loaded onto policy and test rendering.
+TEST_IPV4_ONLY = [nacaddr.IP('10.2.3.4/32')]
+TEST_IPV6_ONLY = [nacaddr.IP('2001:4860:8000::5/128')]
+TEST_IPS = [nacaddr.IP('10.2.3.4/32'), nacaddr.IP('2001:4860:8000::5/128')]
 
 HEADER_TEMPLATE = """
 header {
@@ -179,6 +185,33 @@ def __init__(self, in_dict: dict):
 }
 """
 
+# Input interface name test term.
+SOURCE_INTERFACE_TERM = """
+term src-interface-term {
+  source-interface:: eth123
+  protocol:: tcp
+  action:: accept
+}
+"""
+
+# Output interface name test term.
+DESTINATION_INTERFACE_TERM = """
+term dst-interface-term {
+  destination-interface:: eth123
+  protocol:: tcp
+  action:: accept
+}
+"""
+
+BAD_INTERFACE_TERM = """
+term dst-interface-term {
+  source-interface:: eth123
+  destination-interface:: eth123
+  protocol:: tcp
+  action:: accept
+}
+"""
+
 ESTABLISHED_OPTION_TERM = """
 term established-term {
   protocol:: udp
@@ -233,6 +266,28 @@ def __init__(self, in_dict: dict):
 }
 """
 
+LOGGING_TERM = """
+term log-packets {
+  logging:: true
+  action:: accept
+}
+"""
+
+COUNTER_TERM = """
+term count-packets {
+  counter:: thisnameisignored
+  action:: accept
+}
+"""
+
+COUNT_AND_LOG_TERM = """
+term count-and-log-packets {
+  logging:: true
+  counter:: thisnameisignored
+  action:: accept
+}
+"""
+
 GOOD_TERM_1 = """
 term good-term-1 {
   action:: accept
@@ -248,8 +303,31 @@ def __init__(self, in_dict: dict):
 }
 """
 
-IPV6_TERM_2 = """
-term inet6-icmp {
+IPV6_ONLY_TERM = """
+term ip6-only {
+  destination-address:: TEST_IPV6_ONLY
+  action:: accept
+}
+"""
+
+IPV6_SRCIP = """
+term ip6-src-addr {
+  source-address:: TEST_IPV6_ONLY
+  action:: deny
+}
+"""
+
+IPV4_SRCIP = """
+term ip4-src-addr {
+  source-address:: TEST_IPV4_ONLY
+  action:: deny
+}
+"""
+
+ALL_SRCIP = """
+term all-src-addr {
+  comment:: "All IP address families. v4/v6"
+  source-address:: TEST_IPS
   action:: deny
 }
 """
@@ -260,10 +338,6 @@ def __init__(self, in_dict: dict):
 # This is normally passed from command line.
 EXP_INFO = 2
 
-TEST_IPV4 = nacaddr.IP('10.2.3.4/32')
-TEST_IPV6 = nacaddr.IP('2001:4860:8000::5/128')
-TEST_IPS = [TEST_IPV4, TEST_IPV6]
-
 def IPhelper(addresses):
   """Helper for string to nacaddr.IP conversion for parametized tests."""
   normalized = []
@@ -334,18 +408,24 @@ def testCreateAnonymousSet(self, input_data, expected):
     self.assertEqual(result, expected)
 
   @parameterized.parameters(
-      (['ip6 saddr 2606:4700:4700::1111/128 ip6 daddr { 2001:4860:4860::8844/128, 2001:4860:4860::8888/128 }'], ['tcp sport 80 tcp dport 80'],'ct state { ESTABLISHED, RELATED } log prefix "combo_cnt_log_established" counter',
-       'accept', 'comment ', ['ip6 saddr 2606:4700:4700::1111/128 ip6 daddr { 2001:4860:4860::8844/128, 2001:4860:4860::8888/128 } tcp sport 80 tcp dport 80 ct state { ESTABLISHED, RELATED } log prefix "combo_cnt_log_established" counter accept'
+      ('',['ip6 saddr 2606:4700:4700::1111/128 ip6 daddr { 2001:4860:4860::8844/128, 2001:4860:4860::8888/128 }'], ['tcp sport 80 tcp dport 80'],'ct state { ESTABLISHED, RELATED } log prefix "combo_cnt_log_established" counter',
+       'accept', '', ['ip6 saddr 2606:4700:4700::1111/128 ip6 daddr { 2001:4860:4860::8844/128, 2001:4860:4860::8888/128 } tcp sport 80 tcp dport 80 ct state { ESTABLISHED, RELATED } log prefix "combo_cnt_log_established" counter accept'
        ]),
-      (['ip daddr 8.8.8.8/32'], ['tcp sport 53 tcp dport 53'],'ct state new','accept', 'comment "this is a term with a comment"', ['ip daddr 8.8.8.8/32 tcp sport 53 tcp dport 53 ct state new accept comment "this is a term with a comment"'])
+      ('',['ip daddr 8.8.8.8/32'], ['tcp sport 53 tcp dport 53'],'ct state new','accept', 'comment "this is a term with a comment"', ['ip daddr 8.8.8.8/32 tcp sport 53 tcp dport 53 ct state new accept comment "this is a term with a comment"'])
       )
-  def testGroupExpressions(self, address_expr, porst_proto_expr, opt,
+  def testGroupExpressions(self, int_str, address_expr, porst_proto_expr, opt,
                            verdict, comment, expected_output):
-    result = self.dummyterm.GroupExpressions(address_expr, porst_proto_expr,
+    result = self.dummyterm.GroupExpressions(int_str, address_expr, porst_proto_expr,
                                              opt, verdict, comment)
-
     self.assertEqual(result, expected_output)
 
+  def testBadInterfaceTerm(self):
+    pol = policy.ParsePolicy(GOOD_HEADER_1 + GOOD_TERM_1 + BAD_INTERFACE_TERM,
+                             self.naming)
+    with self.assertRaises(nftables.TermError):
+      nftables.Nftables.__init__(
+          nftables.Nftables.__new__(nftables.Nftables), pol, EXP_INFO)
+
   def testDuplicateTerm(self):
     pol = policy.ParsePolicy(GOOD_HEADER_1 + GOOD_TERM_1 + GOOD_TERM_1,
                              self.naming)
@@ -409,7 +489,7 @@ def testGoodHeader(self):
     nft = str(
         nftables.Nftables(
             policy.ParsePolicy(
-                GOOD_HEADER_1 + GOOD_TERM_1 + GOOD_HEADER_2 + IPV6_TERM_2,
+                GOOD_HEADER_1 + GOOD_TERM_1 + GOOD_HEADER_2 + IPV6_SRCIP,
                 self.naming), EXP_INFO))
     self.assertIn('type filter hook input', nft)
 
@@ -419,7 +499,7 @@ def testStatefulFirewall(self):
     nft = str(
         nftables.Nftables(
             policy.ParsePolicy(
-                GOOD_HEADER_1 + GOOD_TERM_1 + GOOD_HEADER_2 + IPV6_TERM_2,
+                GOOD_HEADER_1 + GOOD_TERM_1 + GOOD_HEADER_2 + IPV6_SRCIP,
                 self.naming), EXP_INFO))
     self.assertIn('ct state established,related accept', nft)
 
@@ -578,7 +658,7 @@ def testRulesetGeneratorICMPmismatch(self, pol_data, doesnotcontain):
     nft = str(
         nftables.Nftables(
             policy.ParsePolicy(pol_data, self.naming), EXP_INFO))
-    self.assertNotRegex(nft, doesnotcontain)
+    self.assertNotIn(doesnotcontain, nft)
 
   def testRulesetGeneratorUniqueChain(self):
     # This test is intended to verify that on mixed address family rulesets
@@ -621,9 +701,9 @@ def testRulesetGeneratorAF(self, policy_data: str, expected_inet: str):
         self.assertNotEmpty(ruleset_list)
         for ruleset in ruleset_list:
           if expected_inet == 'inet':
-            self.assertNotIn(str(TEST_IPV6), ruleset)
+            self.assertNotIn(str(TEST_IPV6_ONLY), ruleset)
           elif expected_inet == 'inet6':
-            self.assertNotIn(str(TEST_IPV4), ruleset)
+            self.assertNotIn(str(TEST_IPV4_ONLY), ruleset)
 
           for rule in ruleset.split('\n'):
             if rule.startswith('ip '):
@@ -633,6 +713,21 @@ def testRulesetGeneratorAF(self, policy_data: str, expected_inet: str):
               self.assertNotIn('ip protocol', rule)
               self.assertNotIn('icmp', rule)
 
+  @parameterized.parameters(
+      (GOOD_HEADER_1 + SOURCE_INTERFACE_TERM, TEST_IPS, '    iifname eth123 meta l4proto'),
+      (GOOD_HEADER_1 + DESTINATION_INTERFACE_TERM, TEST_IPS, '    oifname eth123 meta l4proto'),
+      (GOOD_HEADER_1 + LOGGING_TERM, TEST_IPS, 'log prefix "log-packets"'),
+      (GOOD_HEADER_1 + COUNTER_TERM, TEST_IPS, 'counter'),
+      (GOOD_HEADER_1 + COUNT_AND_LOG_TERM, TEST_IPS, 'log prefix "count-and-log-packets" counter'),
+      (HEADER_MIXED_AF + IPV6_ONLY_TERM, TEST_IPS, 'ip6 daddr 2001:4860:8000::5/128 ct state new accept'),
+      (HEADER_MIXED_AF + ALL_SRCIP, TEST_IPS, 'ip saddr 10.2.3.4/32 drop comment "All IP address families. v4/v6"'),
+  )
+  def testRulesetGenerator(self, policy_data: str, IPs, contains: str):
+    self.naming.GetNetAddr.return_value = IPs
+    nft = str(
+        nftables.Nftables(
+            policy.ParsePolicy(policy_data, self.naming), EXP_INFO))
+    self.assertIn(contains, nft)
 
 if __name__ == '__main__':
   absltest.main()