Merge pull request #215 from da-z97/master

feat: add basic fortiproxy functionality

Author: neubi4

CONTRIBUTING.md

@@ -13,15 +13,15 @@ Setting up the local enviroment is done with [python poetry](https://python-poet
 After making code changes, please run the linters and fix all errors:
 
 ```
-> poetry run tox --elinter
+> poetry run tox -elinter
 ```
 
 ### Auto formatting code
 
 After making code changes, please run code formatters:
 
 ```
-> poetry runtox --eformatter
+> poetry run tox -eformatter
 ```
 
 

fortilib/address.py

@@ -66,7 +66,9 @@ def render(self) -> dict:
             "type": "ipmask",
             "subnet": f"{self.subnet.network_address} {self.subnet.netmask}",
             "comment": self.comment,
-            "associated-interface": self.interface.name if self.interface else "",
+            "associated-interface": (
+                self.interface.name if self.interface else ""
+            ),
             "color": self.color,
         }
 
@@ -115,7 +117,9 @@ def render(self) -> dict:
             "start-ip": str(self.ip_start),
             "end-ip": str(self.ip_end),
             "comment": self.comment,
-            "associated-interface": self.interface.name if self.interface else "",
+            "associated-interface": (
+                self.interface.name if self.interface else ""
+            ),
             "color": self.color,
         }
 
@@ -156,6 +160,8 @@ def render(self) -> dict:
             "type": "fqdn",
             "fqdn": self.fqdn,
             "comment": self.comment,
-            "associated-interface": self.interface.name if self.interface else "",
+            "associated-interface": (
+                self.interface.name if self.interface else ""
+            ),
             "color": self.color,
         }

fortilib/fortiproxy.py

@@ -0,0 +1,224 @@
+from typing import List
+
+from fortilib.firewall import FortigateFirewall
+from fortilib.fortigateapi import (
+    FortiGateApiPolicyDirection,
+    FortigateFirewallApi,
+)
+from fortilib.proxypolicy import FortiproxyPolicy
+
+
+class Fortiproxy(FortigateFirewall):
+    """Fortigate Firewall object.
+    Representation of all firewall objects.
+
+    :param fortigateapi: Fortigate API object (:class:`fortilib.fortigateapi.FortigateFirewallApi`) for API Querys
+    :ivar name: Name of fortigate firewall
+    :ivar fortigate: Fortigate API object (:class:`fortilib.fortigateapi.FortigateFirewallApi`) for API Querys
+    :ivar interfaces: List of :class:`fortilib.interface.FortigateInterface` (default: [])
+    :ivar static_routes: List of :class:`fortilib.routes.FortigateStaticRoute` (default: [])
+    :ivar addresses: List of :class:`fortilib.address.FortigateAddress` (default: [])
+    :ivar address_groups: List of :class:`fortilib.addressgroup.FortigateAddressGroup` (default: [])
+    :ivar vips: List of :class:`fortilib.vip.FortigateVIP` (default: [])
+    :ivar vip_groups: List of :class:`fortilib.vipgroup.FortigateVIPGroup` (default: [])
+    :ivar services: List of :class:`fortilib.service.FortigateService` (default: [])
+    :ivar service_groups: List of :class:`fortilib.servicegroup.FortigateServiceGroup` (default: [])
+    :ivar ippools: List of :class:`fortilib.ippool.FortigateIPPool` (default: [])
+    :ivar policies: List of :class:`fortilib.policy.FortigatePolicy` (default: [])
+    :ivar proxy_addresses: List of :class:`fortilib.proxyaddresses.FortigateProxyAddress` (default: [])
+    :ivar proxy_address_groups: List of :class:`fortilib.proxyaddressgroup.FortigateProxyAddressGroup` (default: [])
+    :ivar proxy_policies: None
+    :ivar all_addresses: List of :class:`fortilib.proxyaddresses.FortigateProxyAddress` and :class:`fortilib.address.FortigateAddress` (default: [])
+    :ivar phase1_interfaces: None
+    :ivar phase2_interfaces: None
+    """
+
+    def __init__(self, name: str, fortigateapi: FortigateFirewallApi):
+        super().__init__(name, fortigateapi)
+
+        self.policies: List[FortiproxyPolicy] = []
+
+        self.proxy_policies = None
+        self.phase1_interfaces = None
+        self.phase2_interfaces = None
+
+    def get_all_objects(self):
+        """Query Fortiproxy API for all firewall objects of the following list.
+
+        - Interfaces
+        - Static Routes
+        - Addresses
+        - Address Groups
+        - VIPs
+        - VIP Groups
+        - Services
+        - Service Groups
+        - IPPools
+        - Policies
+        - Proxy Addresses
+        - Proxy Address Groups
+        - Addresses and Proxy Addresses
+        """
+        self.interfaces = self.get_interfaces()
+        self.static_routes = self.get_static_routes()
+        self.addresses = self.get_addresses()
+        self.address_groups = self.get_address_groups()
+        self.resolve_address_groups()
+        self.proxy_addresses = self.get_proxy_addresses()
+        self.proxy_address_groups = self.get_proxy_address_groups()
+        self.resolve_proxy_address_groups()
+        self.vips = self.get_vips()
+        self.vip_groups = self.get_vip_groups()
+        self.resolve_vip_groups()
+        self.services = self.get_services()
+        self.service_groups = self.get_service_groups()
+        self.resolve_service_groups()
+        self.ippools = self.get_ippools()
+        self.policies = self.get_policies()
+        self.all_addresses = self.get_all_addresses()
+
+    def get_policies(self) -> List[FortiproxyPolicy]:
+        """Query Fortiproxy API for policies
+        :obj:`fortilib.policy.FortigatePolicy` and create list.
+        """
+        policies: List[FortiproxyPolicy] = []
+        for raw in self.fortigate.get_firewall_policies():
+            policy = FortiproxyPolicy.from_dict(raw)
+
+            policy.find_interfaces(self.interfaces)
+            policy.find_addresses(
+                [
+                    self.addresses,
+                    self.address_groups,
+                    self.vips,
+                    self.vip_groups,
+                    self.proxy_addresses,
+                    self.proxy_address_groups,
+                ]
+            )
+            policy.find_services([self.services, self.service_groups])
+
+            policies.append(policy)
+
+        return policies
+
+    def create_firewall_policy(self, policy: FortiproxyPolicy):
+        """Create policy on fortigate with given :obj:`fortilib.proxypolicy.FortiproxyPolicy`."""
+        status = self.fortigate.create_firewall_policy(
+            int(policy.policyid), policy.render()
+        )
+
+        policy.policyid = status["mkey"]
+
+        self.policies.append(policy)
+        return status
+
+    def update_firewall_policy(self, policy: FortiproxyPolicy):
+        """Update policy on fortiproxy with given :obj:`fortilib.proxypolicy.FortiproxyPolicy`."""
+        return self.fortigate.update_firewall_policy(
+            policy.policyid, policy.render()
+        )
+
+    def delete_firewall_policy(self, policy: FortiproxyPolicy):
+        """Delete policy on fortiproxy with given :obj:`fortilib.proxypolicy.FortiproxyPolicy`."""
+        status = self.fortigate.delete_firewall_policy(policy.policyid)
+        self.policies.remove(policy)
+        return status
+
+    def move_firewall_policy(
+        self,
+        policy: FortiproxyPolicy,
+        move_direction: FortiGateApiPolicyDirection,
+        move_identifier_policy: FortiproxyPolicy,
+    ):
+        """move policy on fortiproxy to another position with given :obj:`fortilib.proxypolicy.FortiproxyPolicy`, :obj:`fortilib.fortigateapi.FortiGateApiPolicyDirection` and :obj:`fortilib.policy.FortigatePolicy`."""
+
+        return self.fortigate.move_firewall_policy(
+            policy.policyid,
+            move_direction,
+            move_identifier_policy.policyid,
+        )
+
+    def get_proxy_policies(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.get_proxy_policies.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.get_proxy_policies.__name__}",
+        )
+
+    def update_firewall_proxy_policy(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.update_firewall_proxy_policy.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.update_firewall_proxy_policy.__name__}",
+        )
+
+    def delete_firewall_proxy_policy(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.delete_firewall_proxy_policy.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.delete_firewall_proxy_policy.__name__}",
+        )
+
+    def move_firewall_proxy_policy(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.move_firewall_proxy_policy.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.move_firewall_proxy_policy.__name__}",
+        )
+
+    def get_phase1_interfaces(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.get_phase1_interfaces.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.get_phase1_interfaces.__name__}",
+        )
+
+    def create_firewall_phase1_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.create_firewall_phase1_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.create_firewall_phase1_interface.__name__}",
+        )
+
+    def update_firewall_phase1_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.update_firewall_phase1_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.update_firewall_phase1_interface.__name__}",
+        )
+
+    def delete_firewall_phase1_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.delete_firewall_phase1_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.delete_firewall_phase1_interface.__name__}",
+        )
+
+    def get_phase2_interfaces(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.get_phase2_interfaces.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.get_phase2_interfaces.__name__}",
+        )
+
+    def create_firewall_phase2_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.create_firewall_phase2_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.create_firewall_phase2_interface.__name__}",
+        )
+
+    def update_firewall_phase2_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.update_firewall_phase2_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.update_firewall_phase2_interface.__name__}",
+        )
+
+    def delete_firewall_phase2_interface(self):
+        raise AttributeError(
+            f"{Fortiproxy} has no Method {self.delete_firewall_phase2_interface.__name__}",
+            obj=Fortiproxy,
+            name=f"{self.delete_firewall_phase2_interface.__name__}",
+        )

fortilib/proxypolicy.py

@@ -295,3 +295,48 @@ def render(self) -> dict:
             "profile-group": self.profile_group,
             "comments": self.comment,
         }
+
+
+class FortiproxyPolicy(FortigateProxyPolicy):
+    def __init__(self):
+        super().__init__()
+
+        self.proxy = None
+        self.type = "transparent"
+
+    def populate(self, object_data: dict):
+        super().populate(object_data)
+
+        self.policyid = object_data.get("policyid", self.policyid)
+        self.action = object_data.get("action", self.action)
+        self.status = object_data.get("status", self.status)
+
+        self.schedule = object_data.get("schedule", self.schedule)
+        self.type = object_data.get("type", self.type)
+        self.logtraffic = object_data.get("logtraffic", self.logtraffic)
+        self.utm_status = object_data.get("utm-status", self.utm_status)
+        self.profile_type = object_data.get("profile-type", self.profile_type)
+        self.profile_group = object_data.get(
+            "profile-group", self.profile_group
+        )
+        self.comment = object_data.get("comments", self.comment)
+
+    def render(self) -> dict:
+        return {
+            "policyid": self.policyid,
+            "name": self.name,
+            "type": self.type,
+            "srcintf": get_fortigate_member_array(self.srcintf),
+            "dstintf": get_fortigate_member_array(self.dstintf),
+            "srcaddr": get_fortigate_member_array(self.srcaddr),
+            "dstaddr": get_fortigate_member_array(self.dstaddr),
+            "service": get_fortigate_member_array(self.service),
+            "action": self.action,
+            "status": self.status,
+            "schedule": self.schedule,
+            "logtraffic": self.logtraffic,
+            "utm-status": self.utm_status,
+            "profile-type": self.profile_type,
+            "profile-group": self.profile_group,
+            "comments": self.comment,
+        }