Merge remote-tracking branch 'upstream/master' into issue-295

capirca/lib/cisco.py

@@ -963,6 +963,11 @@ def _TranslatePolicy(self, pol, exp_info):
         filter_options.remove('noverbose')
         self.verbose = False
 
+      self.remove_duplicate_network_objectgroups = False
+      if 'remove_duplicate_network_objectgroups' in filter_options:
+        filter_options.remove('remove_duplicate_network_objectgroups')
+        self.remove_duplicate_network_objectgroups = True
+
       # extended is the most common filter type.
       filter_type = 'extended'
       if len(filter_options) > 1:
@@ -1086,6 +1091,174 @@ def _GetObjectGroupTerm(self, term, filter_name, af=4, verbose=True):
     """Returns an ObjectGroupTerm object."""
     return ObjectGroupTerm(term, filter_name, af=af, verbose=verbose)
 
+  def _remove_duplicate_objects(self, target):
+    """Remove all duplicate object-groups and rename to the first group found.
+
+    Args:
+      target: A string to remove duplicate object-groups from.
+
+    Returns:
+      A string with duplicate object-groups removed.
+    Example:
+    Remove all duplicate object-group and rename to the first group found
+    where the format looks like:
+    object-group network ipv4 firstgroup
+     10.1.1.0/8
+    exit
+    object-group network ipv4 secondgroup
+     10.1.1.0/8
+    exit
+    ipv4 access-list myacl
+     permit tcp net-group firstgroup net-group secondgroup port-group 443-443
+    exit
+    Result:
+    object-group network ipv4 sa_myfirstgroup
+     10.1.1.0/8
+    exit
+    ipv4 access-list myacl
+     permit tcp net-group firstgroup net-group secondgroup port-group 443-443
+    exit
+    """
+    for address_family in ['ipv4', 'ipv6']:
+      duplicate_object_groups = self._get_duplicate_object_groups(
+          target, address_family
+      )
+      target = self._remove_object_groups(
+          target, duplicate_object_groups.keys(), address_family
+      )
+      target = self._replace_object_groups_references(
+          target, duplicate_object_groups, address_family
+      )
+    return target
+
+  def _get_duplicate_object_groups(self, text, address_family):
+    """Parses the given text to extract all object-groups.
+
+    Args:
+      text: A multi-line string with multiple object-groups contained within it.
+      address_family: A string "ipv4" or "ipv6" to remove object-groups of.
+
+    Returns:
+      A dictionary of any duplicate content object-groups with the key of the
+      duplicate name and the value of the first occurrence found name.
+    """
+
+    object_groups = {}
+    duplicate_object_groups = {}
+
+    inside_networkobject = False
+    object_group_name = ''
+    object_group_content = []
+    for line in text.splitlines():
+      if line.startswith('object-group'):
+        fields = line.split()
+        if len(fields) == 4:
+          object_group_type = fields[1]
+          object_group_af = fields[2]
+          object_group_name = fields[3]
+          if (
+              object_group_type == 'network'
+              and object_group_af == address_family
+          ):
+            inside_networkobject = True
+      elif line.startswith('exit') and inside_networkobject:
+        inside_networkobject = False
+        if '\n'.join(object_group_content) in object_groups.keys():
+          duplicate_object_groups[object_group_name] = object_groups[
+              '\n'.join(object_group_content)
+          ]
+        else:
+          object_groups['\n'.join(object_group_content)] = object_group_name
+        object_group_content = []
+      else:
+        if inside_networkobject:
+          object_group_content.append(line)
+    return duplicate_object_groups
+
+  def _remove_object_groups(self, text, objgroups, address_family):
+    """Parses the given text to remove all object-groups asked to be removed.
+
+    Args:
+      text: A multi-line string with multiple object-groups contained within it.
+      objgroups: A set of object-groups to remove.
+      address_family: A string "ipv4" or "ipv6" to remove object-groups of.
+
+    Returns:
+      The original text with the duplicate object-groups removed.
+    """
+
+    inside_networkobject = False
+    skip_line = False
+    return_lines = []
+    for line in text.splitlines():
+      if line.startswith('object-group'):
+        fields = line.split()
+        if len(fields) == 4:
+          object_group_type = fields[1]
+          object_group_af = fields[2]
+          object_group_name = fields[3]
+          if (
+              object_group_type == 'network'
+              and object_group_af == address_family
+          ):
+            inside_networkobject = True
+            if object_group_name in objgroups:
+              skip_line = True
+          else:
+            skip_line = False
+        else:
+          skip_line = False
+      elif line.startswith('exit') and inside_networkobject:
+        inside_networkobject = False
+        if skip_line:
+          skip_line = False
+          continue
+        skip_line = False
+      if skip_line:
+        continue
+      return_lines.append(line)
+    return '\n'.join(return_lines)
+
+  def _replace_object_groups_references(
+      self, text, objgroups, address_family
+  ):
+    """Parses the given text to replace all object-groups in an ACL based on a dictionary.
+
+    Args:
+      text: A multi-line string with ACLs contained within it.
+      objgroups: A dict of object-groups to the key with the value of in
+        references.
+      address_family: A string "ipv4" or "ipv6" to remove object-groups of.
+
+    Returns:
+      The original text with the object-group references changed to those in the
+      dictionary.
+    """
+
+    inside = ''
+    return_lines = []
+    for line in text.splitlines():
+      if line.startswith('ip access-list extended ') or line.startswith(
+          'ipv4 access-list '
+      ):
+        inside = 'ipv4'
+      elif line.startswith('ipv6 access-list ') or line.startswith(
+          'ipv6 access-list '
+      ):
+        inside = 'ipv6'
+      elif line.startswith('exit'):
+        inside = ''
+      elif address_family == inside:
+        fields = line.split()
+        if len(fields) >= 2:
+          for i in range(len(fields) - 1):
+            if fields[i] == 'net-group':
+              if fields[i + 1] in objgroups:
+                fields[i + 1] = objgroups[fields[i + 1]]
+          line = ' ' + ' '.join(fields)
+      return_lines.append(line)
+    return '\n'.join(return_lines)
+
   def _AppendTargetByFilterType(self, filter_name, filter_type):
     """Takes in the filter name and type and appends headers.
 
@@ -1173,4 +1346,6 @@ def __str__(self):
       # ensure that the header is always first
       target = target_header + target
       target += ['', 'exit', '']
+    if self.remove_duplicate_network_objectgroups:
+      return self._remove_duplicate_objects('\n'.join(target))
     return '\n'.join(target)

tests/lib/cisco_test.py

@@ -16,9 +16,9 @@
 
 import datetime
 import re
-from absl.testing import absltest
 from unittest import mock
 
+from absl.testing import absltest
 from capirca.lib import aclgenerator
 from capirca.lib import cisco
 from capirca.lib import nacaddr
@@ -56,6 +56,12 @@
   target:: cisco objgroupheader object-group
 }
 """
+GOOD_OBJGRP_DEDUP_HEADER = """
+header {
+  comment:: "obj group header test"
+  target:: cisco objgroupheader object-group remove_duplicate_network_objectgroups
+}
+"""
 GOOD_OBJGRP_HEADER_2 = """
 header {
   comment:: "obj group header test"
@@ -198,6 +204,15 @@
   action:: accept
 }
 """
+GOOD_TERM_2_DUPE = """
+term good-term-2-dupe {
+  protocol:: tcp
+  destination-address:: SOME_HOST2
+  source-port:: HTTP
+  option:: established
+  action:: accept
+}
+"""
 GOOD_TERM_3 = """
 term good-term-3 {
   protocol:: tcp
@@ -663,12 +678,81 @@ def testObjectGroup(self):
 
     # There should be no addrgroups that look like IP addresses.
     for addrgroup in re.findall(r'net-group ([a-f0-9.:/]+)', str(acl)):
-      self.assertRaises(ValueError, nacaddr.IP(addrgroup))
+      self.assertRaises(ValueError, nacaddr.IP, addrgroup)
 
     self.naming.GetNetAddr.assert_has_calls([mock.call('SOME_HOST'),
                                              mock.call('SOME_HOST')])
     self.naming.GetServiceByProto.assert_called_once_with('HTTP', 'tcp')
 
+  def testObjectGroupNoDuplicates(self):
+    ip_grp1 = ['object-group network ipv4 SOME_HOST']
+    ip_grp1.append(' 10.0.0.0/8')
+    ip_grp1.append('exit')
+    ip_grp2 = ['object-group network ipv4 SOME_HOST2']
+    ip_grp2.append(' 10.0.0.0/8')
+    ip_grp2.append('exit')
+    port_grp1 = ['object-group port 80-80']
+    port_grp1.append(' eq 80')
+    port_grp1.append('exit')
+    port_grp2 = ['object-group port 1024-65535']
+    port_grp2.append(' range 1024 65535')
+    port_grp2.append('exit')
+
+    self.naming.GetNetAddr.side_effect = [
+        [nacaddr.IP('10.0.0.0/8', token='SOME_HOST')],
+        [nacaddr.IP('10.0.0.0/8', token='SOME_HOST2')],
+        [nacaddr.IP('10.0.0.0/8', token='SOME_HOST')],
+        [nacaddr.IP('10.0.0.0/8', token='SOME_HOST2')],
+    ]
+    self.naming.GetServiceByProto.return_value = ['80']
+
+    pol = policy.ParsePolicy(
+        GOOD_OBJGRP_DEDUP_HEADER + GOOD_TERM_2 + GOOD_TERM_2_DUPE, self.naming
+    )
+    acl = cisco.Cisco(pol, EXP_INFO)
+
+    self.assertIn(
+        '\n'.join(ip_grp1), str(acl), '%s %s' % ('\n'.join(ip_grp1), str(acl))
+    )
+    self.assertNotIn(
+        '\n'.join(ip_grp2), str(acl), '%s %s' % ('\n'.join(ip_grp2), str(acl))
+    )
+    self.assertIn(
+        '\n'.join(port_grp1),
+        str(acl),
+        '%s %s' % ('\n'.join(port_grp1), str(acl)),
+    )
+    self.assertIn(
+        '\n'.join(port_grp2),
+        str(acl),
+        '%s %s' % ('\n'.join(port_grp2), str(acl)),
+    )
+
+    # Object-group terms should use the object groups created.
+    self.assertIn(
+        ' permit tcp any port-group 80-80 net-group SOME_HOST port-group'
+        ' 1024-65535',
+        str(acl),
+        str(acl),
+    )
+
+    # Object-group terms should use the first object group not duplicates.
+    self.assertNotIn(
+        ' permit tcp any port-group 80-80 net-group SOME_HOST2 port-group'
+        ' 1024-65535',
+        str(acl),
+        str(acl),
+    )
+
+    # There should be no addrgroups that look like IP addresses.
+    for addrgroup in re.findall(r'net-group ([a-f0-9.:/]+)', str(acl)):
+      self.assertRaises(ValueError, nacaddr.IP, addrgroup)
+
+    self.naming.GetNetAddr.assert_has_calls(
+        [mock.call('SOME_HOST'), mock.call('SOME_HOST2')]
+    )
+    self.naming.GetServiceByProto.assert_called_with('HTTP', 'tcp')
+
   def testObjectGroupInet6(self):
     ip_grp = ['object-group network ipv6 SOME_HOST']
     ip_grp.append(' 2001::3/128')