pa_api.restapi.rest_resources

  1from functools import wraps
  2from multiprocessing.pool import ThreadPool as Pool
  3from typing import ClassVar
  4
  5# List of existing resources
  6# https://docs.paloaltonetworks.com/pan-os/10-2/pan-os-panorama-api/get-started-with-the-pan-os-rest-api/access-the-rest-api
  7
  8DEFAULT_PARAMS = {
  9    "location": "device-group",
 10}
 11
 12
 13class BasePanoramaResource:
 14    def __init__(
 15        self,
 16        client,
 17        host,
 18        resource_type,
 19        resource,
 20        version="v10.1",
 21        default_params=None,
 22    ):
 23        """
 24        client:
 25        """
 26        self.DEFAULT_PARAMS = default_params or DEFAULT_PARAMS
 27        if not host.startswith("http"):
 28            host = "https://" + host
 29        self._client = client
 30        self._name = f"{resource_type}/{resource}"
 31        self._url = f"{host}/restapi/{version}/{resource_type}/{resource}"
 32
 33    def __str__(self):
 34        return self._name
 35
 36    def __repr__(self):
 37        return str(self)
 38
 39    def get(self, params=None, headers=None, data=None, verify=None):
 40        return self._client.get(
 41            self._url,
 42            params=params,
 43            headers=headers,
 44            data=data,
 45            verify=verify,
 46        )
 47
 48    def delete(self, params=None, headers=None, data=None, verify=None):
 49        return self._client.delete(
 50            self._url,
 51            params=params,
 52            headers=headers,
 53            data=data,
 54            verify=verify,
 55        )
 56
 57
 58class PanoramaResource(BasePanoramaResource):
 59    def get(
 60        self,
 61        device_group=None,
 62        name=None,
 63        inherited=True,
 64        params=None,
 65        headers=None,
 66        data=None,
 67        verify=None,
 68    ):
 69        if params is None:
 70            params = {}
 71        _params = {**self.DEFAULT_PARAMS, **params}
 72        if name is not None:
 73            _params["name"] = name
 74
 75        def func(params):
 76            params = {**_params, **params}
 77            res = self._client.get(
 78                self._url,
 79                params=params,
 80                headers=headers,
 81                data=data,
 82                verify=verify,
 83            )
 84            if not isinstance(res, list):
 85                return []  # Error happened
 86            if not inherited:
 87                res = [r for r in res if r["@device-group"] == r["@loc"]]
 88            return res
 89
 90        if device_group is None:
 91            return func(params)
 92        if isinstance(device_group, str):
 93            device_group = [device_group]
 94        if len(device_group) == 1:
 95            params = {"device-group": device_group}
 96            return func(params)
 97        if len(device_group) > 1:
 98            params_list = [{"device-group": dg} for dg in device_group]
 99            with Pool(len(device_group)) as pool:
100                data = pool.map(func, params_list)
101            return [record for res in data for record in res]
102        return None
103
104    def delete(
105        self,
106        device_group=None,
107        name=None,
108        params=None,
109        headers=None,
110        data=None,
111        verify=None,
112    ):
113        if params is None:
114            params = {}
115        params = {**self.DEFAULT_PARAMS, **params}
116        if device_group is not None:
117            params["device-group"] = device_group
118        if name is not None:
119            params["name"] = name
120        return self._client.delete(
121            self._url,
122            params=params,
123            headers=headers,
124            data=data,
125            verify=verify,
126        )
127
128
129class GroupedResources:
130    def __init__(self, *resources):
131        self._resources = resources
132
133    def _grouped(self, method, *args, **kwargs):
134        params_list = [(getattr(r, method), args, kwargs) for r in self._resources]
135
136        def func(args):
137            f, args, kwargs = args
138            return f(*args, **kwargs)
139
140        with Pool(len(self._resources)) as pool:
141            data = pool.map(func, params_list)
142        return [record for res in data for record in res]
143
144    @wraps(PanoramaResource.get)
145    def get(self, *args, **kwargs):
146        return self._grouped("get", *args, **kwargs)
147
148    # def delete(self, params={}, headers={}, data=None, verify=None):
149    #     return self._client.delete(
150    #         self._url,
151    #         params=params, headers=headers,
152    #         data=data, verify=verify
153    #     )
154
155
156class PanoramaResourceType:
157    def __init__(self, client, resource_type):
158        self._client = client
159        self._name = resource_type
160
161    def __str__(self):
162        return self._name
163
164    def __repr__(self):
165        return str(self)
166
167
168class PanoramaPanoramaResourceType(PanoramaResourceType):
169    resource_type = "Panorama"
170
171    def __init__(self, client, domain, version="v10.1"):
172        resource_type = self.resource_type
173        super().__init__(client, resource_type)
174
175        def resource(name):
176            return BasePanoramaResource(
177                client,
178                domain,
179                resource_type,
180                name,
181                version=version,
182            )
183
184        self.DeviceGroups = resource("DeviceGroups")
185
186
187class PanoramaObjectsResourceType(PanoramaResourceType):
188    """
189    Represent the 'Objects' subsection in the API.
190    """
191
192    resource_type = "Objects"
193
194    def __init__(self, client, domain, version="v10.1"):
195        resource_type = self.resource_type
196        super().__init__(client, resource_type)
197
198        def resource(name):
199            return PanoramaResource(
200                client,
201                domain,
202                resource_type,
203                name,
204                version=version,
205            )
206
207        self.Addresses = resource("Addresses")
208        self.AddressGroups = resource("AddressGroups")
209        self.Regions = resource("Regions")
210        self.Applications = resource("Applications")
211        self.ApplicationGroups = resource("ApplicationGroups")
212        self.ApplicationFilters = resource("ApplicationFilters")
213        self.Services = resource("Services")
214        self.ServiceGroups = resource("ServiceGroups")
215        self.Tags = resource("Tags")
216        self.GlobalProtectHIPObjects = resource("GlobalProtectHIPObjects")
217        self.GlobalProtectHIPProfiles = resource("GlobalProtectHIPProfiles")
218        self.ExternalDynamicLists = resource("ExternalDynamicLists")
219        self.CustomDataPatterns = resource("CustomDataPatterns")
220        self.CustomSpywareSignatures = resource("CustomSpywareSignatures")
221        self.CustomVulnerabilitySignatures = resource("CustomVulnerabilitySignatures")
222        self.CustomURLCategories = resource("CustomURLCategories")
223        self.AntivirusSecurityProfiles = resource("AntivirusSecurityProfiles")
224        self.AntiSpywareSecurityProfiles = resource("AntiSpywareSecurityProfiles")
225        self.VulnerabilityProtectionSecurityProfiles = resource(
226            "VulnerabilityProtectionSecurityProfiles",
227        )
228        self.URLFilteringSecurityProfiles = resource("URLFilteringSecurityProfiles")
229        self.FileBlockingSecurityProfiles = resource("FileBlockingSecurityProfiles")
230        self.WildFireAnalysisSecurityProfiles = resource(
231            "WildFireAnalysisSecurityProfiles",
232        )
233        self.DataFilteringSecurityProfiles = resource("DataFilteringSecurityProfiles")
234        self.DoSProtectionSecurityProfiles = resource("DoSProtectionSecurityProfiles")
235        self.SecurityProfileGroups = resource("SecurityProfileGroups")
236        self.LogForwardingProfiles = resource("LogForwardingProfiles")
237        self.AuthenticationEnforcements = resource("AuthenticationEnforcements")
238        self.DecryptionProfiles = resource("DecryptionProfiles")
239        self.DecryptionForwardingProfiles = resource("DecryptionForwardingProfiles")
240        self.Schedules = resource("Schedules")
241        self.SDWANPathQualityProfiles = resource("SDWANPathQualityProfiles")
242        self.SDWANTrafficDistributionProfiles = resource(
243            "SDWANTrafficDistributionProfiles",
244        )
245
246
247class PanoramaPoliciesResourceType(PanoramaResourceType):
248    """
249    Represent the 'Policies' subsection in the API.
250    """
251
252    resource_type = "Policies"
253
254    def __init__(self, client, domain, version="v10.1"):
255        resource_type = self.resource_type
256        super().__init__(client, resource_type)
257
258        def resource(name):
259            return PanoramaResource(
260                client,
261                domain,  # host
262                resource_type,
263                name,
264                version=version,
265            )
266
267        self.SecurityPreRules = resource("SecurityPreRules")
268        self.SecurityPostRules = resource("SecurityPostRules")
269        self.SecurityRules = GroupedResources(
270            self.SecurityPreRules,
271            self.SecurityPostRules,
272        )
273        self.NATPreRules = resource("NATPreRules")
274        self.NATPostRules = resource("NATPostRules")
275        self.NATRules = GroupedResources(self.NATPreRules, self.NATPostRules)
276        self.QoSPreRules = resource("QoSPreRules")
277        self.QoSPostRules = resource("QoSPostRules")
278        self.QoSRules = GroupedResources(self.QoSPreRules, self.QoSPostRules)
279        self.PolicyBasedForwardingPreRules = resource("PolicyBasedForwardingPreRules")
280        self.PolicyBasedForwardingPostRules = resource("PolicyBasedForwardingPostRules")
281        self.PolicyBasedForwardingRules = GroupedResources(
282            self.PolicyBasedForwardingPreRules,
283            self.PolicyBasedForwardingPostRules,
284        )
285        self.DecryptionPreRules = resource("DecryptionPreRules")
286        self.DecryptionPostRules = resource("DecryptionPostRules")
287        self.DecryptionRules = GroupedResources(
288            self.DecryptionPreRules,
289            self.DecryptionPostRules,
290        )
291        self.TunnelInspectionPreRules = resource("TunnelInspectionPreRules")
292        self.TunnelInspectionPostRules = resource("TunnelInspectionPostRules")
293        self.TunnelInspectionRules = GroupedResources(
294            self.TunnelInspectionPreRules,
295            self.TunnelInspectionPostRules,
296        )
297        self.ApplicationOverridePreRules = resource("ApplicationOverridePreRules")
298        self.ApplicationOverridePostRules = resource("ApplicationOverridePostRules")
299        self.ApplicationOverrideRules = GroupedResources(
300            self.ApplicationOverridePreRules,
301            self.ApplicationOverridePostRules,
302        )
303        self.AuthenticationPreRules = resource("AuthenticationPreRules")
304        self.AuthenticationPostRules = resource("AuthenticationPostRules")
305        self.AuthenticationRules = GroupedResources(
306            self.AuthenticationPreRules,
307            self.AuthenticationPostRules,
308        )
309        self.DoSPreRules = resource("DoSPreRules")
310        self.DoSPostRules = resource("DoSPostRules")
311        self.DoSRules = GroupedResources(self.DoSPreRules, self.DoSPostRules)
312        self.SDWANPreRules = resource("SDWANPreRules")
313        self.SDWANPostRules = resource("SDWANPostRules")
314        self.SDWANRules = GroupedResources(self.SDWANPreRules, self.SDWANPostRules)
315
316
317class PanoramaNetworkResourceType(PanoramaResourceType):
318    """
319    Represent the 'Network' subsection in the API.
320    """
321
322    DEFAULT_PARAMS: ClassVar = {
323        "location": "template",
324    }
325    resource_type = "Network"
326
327    def __init__(self, client, domain, version="v10.1"):
328        resource_type = self.resource_type
329        super().__init__(client, resource_type)
330
331        def resource(name):
332            return PanoramaResource(
333                client,
334                domain,
335                resource_type,
336                name,
337                version=version,
338                default_params=self.DEFAULT_PARAMS,
339            )
340
341        self.EthernetInterfaces = resource("EthernetInterfaces")
342        self.AggregateEthernetInterfaces = resource("AggregateEthernetInterfaces")
343        self.VLANInterfaces = resource("VLANInterfaces")
344        self.LoopbackInterfaces = resource("LoopbackInterfaces")
345        self.TunnelIntefaces = resource("TunnelIntefaces")
346        self.SDWANInterfaces = resource("SDWANInterfaces")
347        self.Zones = resource("Zones")
348        self.VLANs = resource("VLANs")
349        self.VirtualWires = resource("VirtualWires")
350        self.VirtualRouters = resource("VirtualRouters")
351        self.IPSecTunnels = resource("IPSecTunnels")
352        self.GRETunnels = resource("GRETunnels")
353        self.DHCPServers = resource("DHCPServers")
354        self.DHCPRelays = resource("DHCPRelays")
355        self.DNSProxies = resource("DNSProxies")
356        self.GlobalProtectPortals = resource("GlobalProtectPortals")
357        self.GlobalProtectGateways = resource("GlobalProtectGateways")
358        self.GlobalProtectGatewayAgentTunnels = resource(
359            "GlobalProtectGatewayAgentTunnels",
360        )
361        self.GlobalProtectGatewaySatelliteTunnels = resource(
362            "GlobalProtectGatewaySatelliteTunnels",
363        )
364        self.GlobalProtectGatewayMDMServers = resource("GlobalProtectGatewayMDMServers")
365        self.GlobalProtectClientlessApps = resource("GlobalProtectClientlessApps")
366        self.GlobalProtectClientlessAppGroups = resource(
367            "GlobalProtectClientlessAppGroups",
368        )
369        self.QoSInterfaces = resource("QoSInterfaces")
370        self.LLDP = resource("LLDP")
371        self.GlobalProtectIPSecCryptoNetworkProfiles = resource(
372            "GlobalProtectIPSecCryptoNetworkProfiles",
373        )
374        self.IKEGatewayNetworkProfiles = resource("IKEGatewayNetworkProfiles")
375        self.IKECryptoNetworkProfiles = resource("IKECryptoNetworkProfiles")
376        self.MonitorNetworkProfiles = resource("MonitorNetworkProfiles")
377        self.InterfaceManagementNetworkProfiles = resource(
378            "InterfaceManagementNetworkProfiles",
379        )
380        self.ZoneProtectionNetworkProfiles = resource("ZoneProtectionNetworkProfiles")
381        self.QoSNetworkProfiles = resource("QoSNetworkProfiles")
382        self.LLDPNetworkProfiles = resource("LLDPNetworkProfiles")
383        self.SDWANInterfaceProfiles = resource("SDWANInterfaceProfiles")
384
385
386class PanoramaDevicesResourceType(PanoramaResourceType):
387    """
388    Represent the 'Devices' subsection in the API.
389    """
390
391    DEFAULT_PARAMS: ClassVar = {
392        "location": "template",
393    }
394    resource_type = "Device"
395
396    def __init__(self, client, domain, version="v10.1"):
397        resource_type = self.resource_type
398        super().__init__(client, resource_type)
399
400        def resource(name):
401            return PanoramaResource(
402                client,
403                domain,
404                resource_type,
405                name,
406                version=version,
407                default_params=self.DEFAULT_PARAMS,
408            )
409
410        self.VirtualSystems = resource("VirtualSystems")
411
412
413__all__ = [
414    "PanoramaDevicesResourceType",
415    "PanoramaNetworkResourceType",
416    "PanoramaObjectsResourceType",
417    "PanoramaPanoramaResourceType",
418    "PanoramaPoliciesResourceType",
419]