Skip established rules in fortigate generator.

Fortigate is stateful and handles return flows automagically: https://www.fortinet.com/resources/cyberglossary/stateful-firewall

PiperOrigin-RevId: 933409011

Author: donaldjray

capirca/lib/fortigate.py

@@ -764,12 +764,12 @@ def _BuildTokens(self):
                          'icmp_code',
                          'fortigate_application_id'}
 
-    supported_sub_tokens.update({'option': {'tcp-established'},
-                                 # Warning, some of these are mapped
-                                 # differently. See _ACTION_TABLE
-                                 'action': {'accept', 'deny',
-                                            'next', 'reject',
-                                            'reject-with-tcp-rst'}})
+    supported_sub_tokens.update({
+        'option': {'established', 'tcp-established'},
+        # Warning, some of these are mapped
+        # differently. See _ACTION_TABLE
+        'action': {'accept', 'deny', 'next', 'reject', 'reject-with-tcp-rst'},
+    })
 
     return supported_tokens, supported_sub_tokens
 
@@ -841,6 +841,12 @@ def _TranslatePolicy(self, pol, exp_info):
               term.name,
               filter_name)
           continue
+        if set(['established', 'tcp-established']).intersection(term.option):
+          logging.debug(
+              'Skipping established term %s because FortiGate is stateful.',
+              term.name,
+          )
+          continue
         if term.expiration:
           if term.expiration <= exp_info_date:
             logging.info(

tests/lib/fortigate_test.py

@@ -292,7 +292,7 @@ def testServices(self):
     ip_term = self.fmt.format(TERM_TEMPLATE,
                               remove_fields=('dest_port', 'src_port'))
     custom_port_term = self.fmt.format(TERM_TEMPLATE, src_port='WHOIS')
-    #print("\icmp_term=========\n", icmp_term)
+    # print("\icmp_term=========\n", icmp_term)
 
     dest_only_acl = fortigate.Fortigate(policy.ParsePolicy(
         GOOD_HEADER + dest_only_term,
@@ -342,7 +342,7 @@ def testInterfaces(self):
     no_interfaces_term = self.fmt.format(TERM_TEMPLATE,
                                          remove_fields=('src_interface',
                                                         'dest_interface'))
-    #print("no_interfaces_term=", no_interfaces_term)
+    # print("no_interfaces_term=", no_interfaces_term)
     src_only_int_term = self.fmt.format(TERM_TEMPLATE,
                                         src_interface='wan1',
                                         remove_fields=('dest_interface',))
@@ -419,6 +419,22 @@ def testExpiration(self):
         and expiration_sig in str(expiration_acl),
         '[%s]' % str(expiration_acl))
 
+  def testEstablished(self):
+    """Tests that established terms are skipped."""
+    established_term = self.fmt.format(
+        TERM_TEMPLATE,
+        add_fields={'option': 'established', 'comment': '"test established"'},
+    )
+
+    established_acl = fortigate.Fortigate(
+        policy.ParsePolicy(GOOD_HEADER + established_term, self.naming),
+        EXP_INFO,
+    )
+
+    self.assertNotIn(
+        'test established', str(established_acl), '[%s]' % str(established_acl)
+    )
+
   def testApplication_ID(self):
     """
     Tests an application ID being used.