From 9d0c63c3afe655d094252b743938b1123c33661e Mon Sep 17 00:00:00 2001
From: Hannu Kamarainen <hannu.kamarainen@cern.ch>
Date: Wed, 6 Jan 2021 13:19:04 +0100
Subject: [PATCH] Cisco ASA: Add enable_dsmo header option

---
 capirca/lib/ciscoasa.py    | 26 +++++++++++++++++++++++---
 tests/lib/ciscoasa_test.py | 29 +++++++++++++++++++++++++++++
 2 files changed, 52 insertions(+), 3 deletions(-)

diff --git a/capirca/lib/ciscoasa.py b/capirca/lib/ciscoasa.py
index 003105a9..25b2a62d 100644
--- a/capirca/lib/ciscoasa.py
+++ b/capirca/lib/ciscoasa.py
@@ -29,6 +29,7 @@
 from capirca.lib import aclgenerator
 from capirca.lib import cisco
 from capirca.lib import nacaddr
+from capirca.lib import summarizer
 
 
 _ACTION_TABLE = {
@@ -64,12 +65,13 @@ class NoCiscoPolicyError(Error):
 class Term(cisco.Term):
   """A single ACL Term."""
 
-  def __init__(self, term, filter_name, af=4):
+  def __init__(self, term, filter_name, af=4, enable_dsmo=False):
     self.term = term
     self.filter_name = filter_name
     self.options = []
     assert af in (4, 6)
     self.af = af
+    self.enable_dsmo = enable_dsmo
 
   def __str__(self):
     # Verify platform specific terms. Skip whole term if platform does not
@@ -123,6 +125,8 @@ def __str__(self):
         source_address = nacaddr.ExcludeAddrs(
             source_address,
             source_address_exclude)
+      if len(source_address) > 1 and self.enable_dsmo:
+        source_address = summarizer.Summarize(source_address)
     else:
       # source address not set
       source_address = ['any']
@@ -137,6 +141,8 @@ def __str__(self):
         destination_address = nacaddr.ExcludeAddrs(
             destination_address,
             destination_address_exclude)
+      if len(destination_address) > 1 and self.enable_dsmo:
+        destination_address = summarizer.Summarize(destination_address)
     else:
       # destination address not set
       destination_address = ['any']
@@ -184,6 +190,8 @@ def __str__(self):
                        (saddr == 'any')) and
                       ((isinstance(daddr, nacaddr.IPv4)) or (daddr == 'any'))):
                     do_output = True
+                  elif isinstance(saddr, summarizer.DSMNet) or isinstance(daddr, summarizer.DSMNet):
+                    do_output = True
                 if self.af == 6:
                   if (((isinstance(saddr, nacaddr.IPv6)) or
                        (saddr == 'any')) and
@@ -252,6 +260,16 @@ def _TermletToStr(self, filter_name, action, proto, saddr, sport, daddr,
       else:
         daddr = 'host %s' % (daddr.network_address)
 
+    if isinstance(saddr, summarizer.DSMNet):
+      saddr = '%s %s' % summarizer.ToDottedQuad(saddr, negate=False)
+      if saddr.endswith('255.255.255.255'):
+        saddr = f'host {saddr.split()[0]}'
+
+    if isinstance(daddr, summarizer.DSMNet):
+      daddr = '%s %s' % summarizer.ToDottedQuad(daddr, negate=False)
+      if daddr.endswith('255.255.255.255'):
+        daddr = f'host {daddr.split()[0]}'
+
     # fix ports
     if not sport:
       sport = ''
@@ -325,7 +343,8 @@ def _TranslatePolicy(self, pol, exp_info):
     exp_info_date = current_date + datetime.timedelta(weeks=exp_info)
 
     for header, terms in self.policy.filters:
-      filter_name = header.FilterName('ciscoasa')
+      filter_options = header.FilterOptions(self._PLATFORM)
+      filter_name = header.FilterName(self._PLATFORM)
 
       new_terms = []
       # now add the terms
@@ -339,7 +358,8 @@ def _TranslatePolicy(self, pol, exp_info):
                             'will not be rendered.', term.name, filter_name)
             continue
 
-        new_terms.append(str(Term(term, filter_name)))
+        enable_dsmo = len(filter_options) > 1 and 'enable_dsmo' in filter_options[1:]
+        new_terms.append(str(Term(term, filter_name, enable_dsmo=enable_dsmo)))
 
       self.ciscoasa_policies.append((header, filter_name, new_terms))
 
diff --git a/tests/lib/ciscoasa_test.py b/tests/lib/ciscoasa_test.py
index 5cd0e11f..8c003bfc 100644
--- a/tests/lib/ciscoasa_test.py
+++ b/tests/lib/ciscoasa_test.py
@@ -22,6 +22,7 @@
 import unittest
 
 from capirca.lib import ciscoasa
+from capirca.lib import nacaddr
 from capirca.lib import naming
 from capirca.lib import policy
 import mock
@@ -34,6 +35,13 @@
 }
 """
 
+GOOD_DSMO_HEADER = """
+header {
+  comment:: "this is a test acl"
+  target:: ciscoasa test-filter enable_dsmo
+}
+"""
+
 GOOD_TERM_1 = """
 term good-term-1 {
   verbatim:: ciscoasa "mary had a little lamb"
@@ -49,6 +57,14 @@
 }
 """
 
+GOOD_DSMO_TERM = """
+term good-dsmo-term {
+  protocol:: tcp
+  destination-address:: SOME_HOST
+  action:: accept
+}
+"""
+
 SUPPORTED_TOKENS = {
     'action',
     'comment',
@@ -144,6 +160,19 @@ def testBuildWarningTokens(self):
     self.assertEqual(st, SUPPORTED_TOKENS)
     self.assertEqual(sst, SUPPORTED_SUB_TOKENS)
 
+  def testDsmo(self):
+    addr_list = list()
+    for octet in range(0, 256):
+      net = nacaddr.IP('192.168.' + str(octet) + '.64/27')
+      addr_list.append(net)
+    self.naming.GetNetAddr.return_value = addr_list
+
+    acl = ciscoasa.CiscoASA(policy.ParsePolicy(GOOD_DSMO_HEADER + GOOD_DSMO_TERM,
+                                               self.naming), EXP_INFO)
+    self.assertIn('permit tcp any 192.168.0.64 255.255.0.224', str(acl))
+
+    self.naming.GetNetAddr.assert_called_once_with('SOME_HOST')
+
 
 if __name__ == '__main__':
   unittest.main()