fix: dealing with different address types in phase2 interfaces (#121)

Author: BenjaminBoehm

fortilib/__init__.py

@@ -29,3 +29,7 @@ def get_fortigate_member_array(source: List, attrname="name") -> List[Dict]:
         )
 
     return ret
+
+
+def remove_empty_dict_values(source: dict):
+    return {k: v for k, v in source.items() if v}

fortilib/firewall.py

@@ -62,6 +62,8 @@ class FortigateFirewall:
     :ivar proxy_address_groups: List of :class:`fortilib.proxyaddressgroup.FortigateProxyAddressGroup` (default: [])
     :ivar proxy_policies: List of :class:`fortilib.proxypolicies.FortigateProxyPolicy` (default: [])
     :ivar all_addresses: List of :class:`fortilib.proxyaddresses.FortigateProxyAddress` and :class:`fortilib.address.FortigateAddress` (default: [])
+    :ivar phase1_interfaces: List of :class:`fortilib.phase1interface.FortigatePhase1Interface` (default: [])
+    :ivar phase2_interfaces: List of :class:`fortilib.phase2interface.FortigatePhase2Interface`` (default: [])
     """
 
     def __init__(self, name: str, fortigateapi: FortigateFirewallApi):

fortilib/phase2interface.py

@@ -1,5 +1,6 @@
 import ipaddress
 
+from fortilib import remove_empty_dict_values
 from fortilib.base import FortigateNamedObject
 
 
@@ -9,12 +10,24 @@ class FortigatePhase2KeylifeType:
     BOTH = "both"
 
 
+class FortigatePhase2AddressType:
+    IP = "ip"
+    RANGE = "range"
+    SUBNET = "subnet"
+
+
 class FortigatePhase2Interface(FortigateNamedObject):
     """Fortigate object for phase2 interfaces.
 
     :ivar phase1_name: Name of phase1
+    :ivar src_start_ip: Source address e.g. 10.0.0.10
+    :ivar src_end_ip: Source address e.g. 10.0.0.20
     :ivar src_subnet: Source subnet e.g. 10.0.0.0/8
+    :ivar src_addr_type: Source address type e.g. subnet
+    :ivar dst_start_ip: Destination address e.g. 192.168.100.10
+    :ivar dst_end_ip: Destination address e.g. 192.168.100.20
     :ivar dst_subnet: Destination subnet e.g. 192.168.100.0/24
+    :ivar dst_addr_type: Destination address type e.g. subnet
     :ivar dhgrp: Diffie-Hellman group e.g. 20
     :ivar auto_negotiate: Auto-negotiate enable/disable (default: "disable")
     :ivar keepalive: Keepalive enable/disable (default: "disable")
@@ -32,6 +45,16 @@ def __init__(self):
         self.phase1_name: str = ""
         self.dst_subnet: ipaddress.IPv4Network = None
         self.src_subnet: ipaddress.IPv4Network = None
+        self.dst_start_ip: ipaddress.IPv4Address = None
+        self.dst_end_ip: ipaddress.IPv4Address = None
+        self.src_start_ip: ipaddress.IPv4Address = None
+        self.src_end_ip: ipaddress.IPv4Address = None
+        self.dst_addr_type: FortigatePhase2AddressType = (
+            FortigatePhase2AddressType.SUBNET
+        )
+        self.src_addr_type: FortigatePhase2AddressType = (
+            FortigatePhase2AddressType.SUBNET
+        )
         self.dhgrp: str = ""
         self.auto_negotiate: str = "disable"
         self.keepalive: str = "disable"
@@ -58,18 +81,50 @@ def populate(self, object_data: dict):
         super().populate(object_data)
 
         self.phase1_name = object_data.get("phase1name", self.phase1_name)
-        self.dst_subnet = ipaddress.ip_network(
-            "{}/{}".format(
-                object_data.get("dst-subnet", "0.0.0.0/0").split()[0],
-                object_data.get("dst-subnet", "0.0.0.0/0").split()[1],
-            )
+        self.dst_addr_type = object_data.get(
+            "dst-addr-type", FortigatePhase2AddressType.SUBNET
         )
-        self.src_subnet = ipaddress.ip_network(
-            "{}/{}".format(
-                object_data.get("src-subnet", "0.0.0.0/0").split()[0],
-                object_data.get("src-subnet", "0.0.0.0/0").split()[1],
-            )
+        self.src_addr_type = object_data.get(
+            "src-addr-type", FortigatePhase2AddressType.SUBNET
         )
+        if self.dst_addr_type == "subnet":
+            self.dst_subnet = ipaddress.ip_network(
+                "{}/{}".format(
+                    object_data.get("dst-subnet", "0.0.0.0/0").split()[0],
+                    object_data.get("dst-subnet", "0.0.0.0/0").split()[1],
+                )
+            )
+
+        if self.src_addr_type == FortigatePhase2AddressType.SUBNET:
+            self.src_subnet = ipaddress.ip_network(
+                "{}/{}".format(
+                    object_data.get("src-subnet", "0.0.0.0/0").split()[0],
+                    object_data.get("src-subnet", "0.0.0.0/0").split()[1],
+                )
+            )
+        if (
+            self.dst_addr_type == FortigatePhase2AddressType.IP
+            or self.dst_addr_type == FortigatePhase2AddressType.RANGE
+        ):
+            self.dst_start_ip = ipaddress.ip_address(
+                object_data.get("dst-start-ip", "0.0.0.0")
+            )
+            if self.dst_addr_type == FortigatePhase2AddressType.RANGE:
+                self.dst_end_ip = ipaddress.ip_address(
+                    object_data.get("dst-end-ip", "0.0.0.0")
+                )
+        if (
+            self.src_addr_type == FortigatePhase2AddressType.IP
+            or self.src_addr_type == FortigatePhase2AddressType.RANGE
+        ):
+            self.src_start_ip = ipaddress.ip_address(
+                object_data.get("src-start-ip", "0.0.0.0")
+            )
+            if self.src_addr_type == FortigatePhase2AddressType.RANGE:
+                self.src_end_ip = ipaddress.ip_address(
+                    object_data.get("src-end-ip", "0.0.0.0")
+                )
+
         self.dhgrp = object_data.get("dhgrp", self.dhgrp)
         self.auto_negotiate = object_data.get(
             "auto-negotiate", self.auto_negotiate
@@ -94,36 +149,54 @@ def render(self) -> dict:
                 {
                     "name": "vpn_phase2",
                     "phase1name": "vpn_phase1",
+                    "dst-addr-type": "subnet"
                     "dst-subnet": "192.168.100.0/24",
+                    "dst-start-ip": "192.168.100.1",
+                    "dst-end-ip": "192.168.100.128",
+                    "src-addr-type": "subnet"
                     "src-subnet": "10.0.0.0/8",
+                    "src-start-ip": "10.0.0.1",
+                    "src-end-ip": "10.0.0.128",
                     "dhgrp": "20",
-                    "pfs":"enable",
-                    "replay":"enable",
-                    "keepalive":"disable",
-                    "auto-negotiate":"enable",
-                    "keylifeseconds":14400,
-                    "keylifekbs":5120,
-                    "keylife-type":"seconds",
+                    "pfs": "enable",
+                    "replay": "enable",
+                    "keepalive": "disable",
+                    "auto-negotiate": "enable",
+                    "keylifeseconds": 14400,
+                    "keylifekbs": 5120,
+                    "keylife-type": "seconds",
                     "proposal": "chacha20poly1305 aes256-sha512 aes256gcm",
                     "comments": "",
                 },
         """
-        return {
-            "name": self.name,
-            "phase1name": self.phase1_name,
-            "dst-subnet": str(self.dst_subnet),
-            "src-subnet": str(self.src_subnet),
-            "dhgrp": self.dhgrp,
-            "pfs": self.pfs,
-            "replay": self.replay,
-            "keepalive": self.keepalive,
-            "auto-negotiate": self.auto_negotiate,
-            "keylifeseconds": self.keylife_seconds,
-            "keylifekbs": self.keylife_kbs,
-            "keylife-type": self.keylife_type,
-            "proposal": self.proposal,
-            "comments": self.comment,
-        }
+        return remove_empty_dict_values(
+            {
+                "name": self.name,
+                "phase1name": self.phase1_name,
+                "dst-addr-type": self.dst_addr_type,
+                "dst-subnet": str(self.dst_subnet) if self.dst_subnet else "",
+                "dst-start-ip": (
+                    str(self.dst_start_ip) if self.dst_start_ip else ""
+                ),
+                "dst-end-ip": str(self.dst_end_ip) if self.dst_end_ip else "",
+                "src-addr-type": self.src_addr_type,
+                "src-subnet": str(self.src_subnet) if self.src_subnet else "",
+                "src-start-ip": (
+                    str(self.src_start_ip) if self.src_start_ip else ""
+                ),
+                "src-end-ip": str(self.src_end_ip) if self.src_end_ip else "",
+                "dhgrp": self.dhgrp,
+                "pfs": self.pfs,
+                "replay": self.replay,
+                "keepalive": self.keepalive,
+                "auto-negotiate": self.auto_negotiate,
+                "keylifeseconds": self.keylife_seconds,
+                "keylifekbs": self.keylife_kbs,
+                "keylife-type": self.keylife_type,
+                "proposal": self.proposal,
+                "comments": self.comment,
+            }
+        )
 
     def __repr__(self):
-        return f"{self.__class__.__name__} {self.name} Phase1 Name: {self.phase1_name} SRC: {self.src_subnet} DST: {self.dst_subnet}"
+        return f"{self.__class__.__name__} {self.name} Phase1 Name: {self.phase1_name}"

tests/data.py

@@ -2556,5 +2556,47 @@
         "dst-end-ip6": "::",
         "dst-subnet": "192.168.100.0 255.255.255.0",
         "dst-port": 0,
-    }
+    },
+    {
+        "name": "vpn_phase2_2",
+        "q_origin_key": "vpn_phase2_2",
+        "phase1name": "vpn_phase1",
+        "dhcp-ipsec": "disable",
+        "proposal": "chacha20poly1305 aes256gcm",
+        "pfs": "enable",
+        "ipv4-df": "disable",
+        "dhgrp": "20",
+        "replay": "enable",
+        "keepalive": "enable",
+        "auto-negotiate": "enable",
+        "add-route": "phase1",
+        "inbound-dscp-copy": "phase1",
+        "auto-discovery-sender": "phase1",
+        "auto-discovery-forwarder": "phase1",
+        "keylifeseconds": 14400,
+        "keylifekbs": 5120,
+        "keylife-type": "seconds",
+        "single-source": "disable",
+        "route-overlap": "use-new",
+        "encapsulation": "tunnel-mode",
+        "l2tp": "disable",
+        "comments": "test phase2",
+        "initiator-ts-narrow": "disable",
+        "diffserv": "disable",
+        "diffservcode": "000000",
+        "protocol": 0,
+        "src-name": "",
+        "src-name6": "",
+        "src-addr-type": "range",
+        "src-start-ip": "10.0.0.10",
+        "src-end-ip": "10.0.0.11",
+        "src-end-ip6": "::",
+        "src-port": 0,
+        "dst-name": "",
+        "dst-name6": "",
+        "dst-addr-type": "ip",
+        "dst-start-ip": "192.168.100.10",
+        "dst-end-ip6": "::",
+        "dst-port": 0,
+    },
 ]

tests/tests_fortilib_firewall_read/test_fortilib_firewall_phase2_interfaces.py

@@ -16,7 +16,9 @@ def test_firewall_base_phase2_interfaces(self):
             {
                 "name": "vpn_phase2",
                 "phase1name": "vpn_phase1",
+                "dst-addr-type": "subnet",
                 "dst-subnet": "192.168.100.0/24",
+                "src-addr-type": "subnet",
                 "src-subnet": "10.0.0.0/8",
                 "dhgrp": "20",
                 "pfs": "enable",
@@ -39,5 +41,31 @@ def test_firewall_base_phase2_interfaces(self):
 
         self.assertEqual(
             str(phase2_interface),
-            "FortigatePhase2Interface vpn_phase2 Phase1 Name: vpn_phase1 SRC: 10.0.0.0/8 DST: 192.168.100.0/24",
+            "FortigatePhase2Interface vpn_phase2 Phase1 Name: vpn_phase1",
+        )
+
+        phase2_interface: FortigatePhase1Interface = get_by(
+            "name", "vpn_phase2_2", self.fw.phase2_interfaces
+        )
+        self.assertEqual(
+            phase2_interface.render(),
+            {
+                "name": "vpn_phase2_2",
+                "phase1name": "vpn_phase1",
+                "dst-addr-type": "ip",
+                "dst-start-ip": "192.168.100.10",
+                "src-addr-type": "range",
+                "src-start-ip": "10.0.0.10",
+                "src-end-ip": "10.0.0.11",
+                "dhgrp": "20",
+                "pfs": "enable",
+                "replay": "enable",
+                "keepalive": "enable",
+                "auto-negotiate": "enable",
+                "keylifeseconds": 14400,
+                "keylifekbs": 5120,
+                "keylife-type": "seconds",
+                "proposal": "chacha20poly1305 aes256gcm",
+                "comments": "test phase2",
+            },
         )

tests/tests_fortilib_firewall_write/test_fortilib_firewall_phase2_interface.py

@@ -26,15 +26,16 @@ def test_firewall_base_phase2_interface_create(self):
             {
                 "name": "vpn_phase2",
                 "phase1name": "vpn_phase1",
+                "dst-addr-type": "subnet",
                 "dst-subnet": "192.168.100.0/24",
+                "src-addr-type": "subnet",
                 "src-subnet": "10.0.0.0/8",
                 "dhgrp": "20",
                 "pfs": "enable",
                 "replay": "enable",
                 "keepalive": "enable",
                 "auto-negotiate": "disable",
                 "keylifeseconds": 14400,
-                "keylifekbs": None,
                 "keylife-type": "seconds",
                 "proposal": "chacha20poly1305 aes256gcm",
                 "comments": "test phase2",
@@ -57,7 +58,9 @@ def test_firewall_base_phase2_interface_update(self):
             {
                 "name": "vpn_phase2",
                 "phase1name": "vpn_phase1",
+                "dst-addr-type": "subnet",
                 "dst-subnet": "192.168.200.0/24",
+                "src-addr-type": "subnet",
                 "src-subnet": "10.0.0.0/8",
                 "dhgrp": "20",
                 "pfs": "enable",