Merge pull request #275 from kevinsteves:br12

PiperOrigin-RevId: 375543842

Author: Capirca Team

capirca/lib/paloaltofw.py

@@ -141,10 +141,10 @@ class ServiceMap(object):
   def __init__(self):
     self.entries = {}
 
-  def get_service_name(self, term_name, ports, protocol):
+  def get_service_name(self, term_name, src_ports, ports, protocol):
     """Returns service name based on the provided ports and protocol."""
-    if (ports, protocol) in self.entries:
-      return self.entries[(ports, protocol)]["name"]
+    if (src_ports, ports, protocol) in self.entries:
+      return self.entries[(src_ports, ports, protocol)]["name"]
 
     service_name = "service-%s-%s" % (term_name, protocol)
 
@@ -158,7 +158,7 @@ def get_service_name(self, term_name, ports, protocol):
             "You have a duplicate service. A service named %s already exists." %
             service_name)
 
-    self.entries[(ports, protocol)] = {"name": service_name}
+    self.entries[(src_ports, ports, protocol)] = {"name": service_name}
     return service_name
 
 
@@ -195,6 +195,16 @@ def ModifyOptions(self, terms, service_map):
     self.options["service"] = []
     self.options["logging"] = []
 
+    def pan_ports(ports):
+      x = []
+      for tup in ports:
+        if len(tup) > 1 and tup[0] != tup[1]:
+          x.append(str(tup[0]) + "-" + str(tup[1]))
+        else:
+          x.append(str(tup[0]))
+
+      return tuple(x)
+
     # COMMENT
     if term.comment:
       self.options["description"] = term.comment
@@ -244,18 +254,17 @@ def ModifyOptions(self, terms, service_map):
       for pan_app in term.pan_application:
         self.options["application"].append(pan_app)
 
-    if term.destination_port:
-      ports = []
-      for tup in term.destination_port:
-        if len(tup) > 1 and tup[0] != tup[1]:
-          ports.append(str(tup[0]) + "-" + str(tup[1]))
-        else:
-          ports.append(str(tup[0]))
-      ports = tuple(ports)
+    if term.source_port or term.destination_port:
+      src_ports = pan_ports(term.source_port)
+      if term.destination_port:
+        ports = pan_ports(term.destination_port)
+      else:
+        ports = pan_ports([("0", "65535")])
 
       # check to see if this service already exists
       for p in term.protocol:
-        service_name = service_map.get_service_name(term.name, ports, p)
+        service_name = service_map.get_service_name(term.name,
+                                                    src_ports, ports, p)
         if service_name not in self.options["service"]:
           self.options["service"].append(service_name)
 
@@ -757,12 +766,20 @@ def __str__(self):
     for k, v in self.service_map.entries.items():
       entry = etree.SubElement(service, "entry", {"name": v["name"]})
       proto0 = etree.SubElement(entry, "protocol")
-      proto = etree.SubElement(proto0, k[1])
+      proto = etree.SubElement(proto0, k[2])
+      # destination port
       port = etree.SubElement(proto, "port")
-      tup = str(k[0])[1:-1]
+      tup = str(k[1])[1:-1]
       if tup[-1] == ",":
         tup = tup[:-1]
       port.text = tup.replace("'", "").replace(", ", ",")
+      # source port
+      if len(k[0]):
+        sport = etree.SubElement(proto, "source-port")
+        tup = str(k[0])[1:-1]
+        if tup[-1] == ",":
+          tup = tup[:-1]
+        sport.text = tup.replace("'", "").replace(", ", ",")
 
     # RULES
     vsys_entry.append(etree.Comment(" Rules "))

tests/lib/paloaltofw_test.py

@@ -440,6 +440,7 @@
 PATH_VSYS = "./devices/entry[@name='localhost.localdomain']/vsys/entry[@name='vsys1']"
 PATH_RULES = PATH_VSYS + '/rulebase/security/rules'
 PATH_TAG = PATH_VSYS + '/tag'
+PATH_SERVICE = PATH_VSYS + '/service'
 
 
 class PaloAltoFWTest(unittest.TestCase):
@@ -472,10 +473,10 @@ def testServiceMap(self):
         policy.ParsePolicy(GOOD_HEADER_1 + SVC_TERM_1, definitions), EXP_INFO)
     self.assertEqual(
         pol1.service_map.entries, {
-            (('22',), 'tcp'): {
+            ((), ('22',), 'tcp'): {
                 'name': 'service-ssh-term-1-tcp'
             },
-            (('25',), 'tcp'): {
+            ((), ('25',), 'tcp'): {
                 'name': 'service-smtp-term-1-tcp'
             }
         }, pol1.service_map.entries)
@@ -485,7 +486,7 @@ def testServiceMap(self):
     # The expectation is that there will be a single port mapped.
     self.assertEqual(
         pol2.service_map.entries, {
-            (('25',), 'tcp'): {
+            ((), ('25',), 'tcp'): {
                 'name': 'service-smtp-term-1-tcp'
             }
         }, pol2.service_map.entries)
@@ -869,6 +870,69 @@ def testPanApplication(self):
       apps = {elem.text for elem in x}
       self.assertEqual(APPS[i], apps, output)
 
+  def testPanPorts(self):
+    POL = """
+header {
+  target:: paloalto from-zone trust to-zone untrust
+}
+term rule-1 {
+%s
+  action:: accept
+}"""
+
+    T = """
+  protocol:: udp
+  destination-port:: NTP
+"""
+
+    definitions = naming.Naming()
+    definitions._ParseLine('NTP = 123/tcp 123/udp', 'services')
+    definitions._ParseLine('DNS = 53/tcp 53/udp', 'services')
+
+    pol = policy.ParsePolicy(POL % T, definitions)
+    paloalto = paloaltofw.PaloAltoFW(pol, EXP_INFO)
+    output = str(paloalto)
+    name = "service-rule-1-udp"
+    path = "/entry[@name='%s']/protocol/udp/port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertEqual(x, "123", output)
+    path = "/entry[@name='%s']/protocol/udp/source-port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertIsNone(x, output)
+
+    T = """
+  protocol:: udp
+  source-port:: NTP
+"""
+
+    pol = policy.ParsePolicy(POL % T, definitions)
+    paloalto = paloaltofw.PaloAltoFW(pol, EXP_INFO)
+    output = str(paloalto)
+    name = "service-rule-1-udp"
+    path = "/entry[@name='%s']/protocol/udp/port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertEqual(x, "0-65535", output)
+    path = "/entry[@name='%s']/protocol/udp/source-port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertEqual(x, "123", output)
+
+    T = """
+  protocol:: tcp
+  source-port:: NTP
+  destination-port:: NTP DNS
+"""
+
+    pol = policy.ParsePolicy(POL % T, definitions)
+    paloalto = paloaltofw.PaloAltoFW(pol, EXP_INFO)
+    output = str(paloalto)
+    name = "service-rule-1-tcp"
+    path = "/entry[@name='%s']/protocol/tcp/port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertEqual(x, "53,123", output)
+    path = "/entry[@name='%s']/protocol/tcp/source-port" % name
+    x = paloalto.config.findtext(PATH_SERVICE + path)
+    self.assertEqual(x, "123", output)
+
 
 if __name__ == '__main__':
   unittest.main()