# Palo Alto in Python

Query traffic, threat and URL logs from Palo Alto Panorama in Python services.

Your services query Palo Alto Panorama logs through an outgoing REST connection. Panorama's API returns XML and runs each query as an asynchronous job - the service submits the query, polls until the job completes and fetches the results.

## Query traffic logs {#query-traffic-logs}

The service below submits a traffic log query, polls until the job finishes and maps the entries into its response:

```python
# -*- coding: utf-8 -*-

import time
import xmltodict
from zato.server.service import Service

class GetTrafficLogs(Service):
    name = 'security.paloalto.traffic-logs'

    input = 'query', '-nlogs'

    def handle(self):

        # Connection defined in Dashboard
        conn = self.rest['Panorama']

        # Default to 10 logs if not specified
        nlogs = self.request.input.nlogs
        if not nlogs:
            nlogs = 10

        # Headers for authentication
        headers = {'Authorization': 'Basic YOUR_BASE64_CREDENTIALS'}

        # Submit the log query
        params = {
            'type': 'log',
            'log-type': 'traffic',
            'query': self.request.input.query,
            'nlogs': nlogs
        }

        response = conn.get(params, headers=headers)
        result = xmltodict.parse(response.data)

        # Get the job ID
        job_id = result['response']['result']['job']

        # Poll until complete
        job_status = ''
        attempts = 0

        while job_status != 'FIN' and attempts < 40:
            time.sleep(3)
            attempts += 1

            status_params = {
                'type': 'log',
                'action': 'get',
                'job-id': job_id
            }

            status_response = conn.get(status_params, headers=headers)
            status_result = xmltodict.parse(status_response.data)
            job_status = status_result['response']['result']['job']['status']

        if job_status != 'FIN':
            raise Exception('Query timed out')

        # Fetch the results
        logs = status_result['response']['result']['log']['logs']
        count = logs['@count']
        entries = logs.get('entry', [])

        self.response.payload = {
            'count': count,
            'entries': self._map_entries(entries)
        }

    def _map_entries(self, entries):

        if not entries:
            return []

        # Handle single entry (not a list)
        if not isinstance(entries, list):
            entries = [entries]

        return [
            {
                'action': e['action'],
                'app': e['app'],
                'src': e['src'],
                'dst': e['dst'],
                'port': e['dport'],
                'time': e['receive_time']
            }
            for e in entries
        ]
```

## Query threat logs {#query-threat-logs}

Threat logs record security incidents - the query defaults to everything except file subtypes:

```python
# -*- coding: utf-8 -*-

import xmltodict
from zato.server.service import Service

class GetThreatLogs(Service):
    name = 'security.paloalto.threat-logs'

    input = '-query', '-nlogs'

    def handle(self):

        conn = self.rest['Panorama']
        headers = {'Authorization': 'Basic YOUR_BASE64_CREDENTIALS'}

        # Default query for recent threats
        query = self.request.input.query
        if not query:
            query = '(subtype neq file)'

        nlogs = self.request.input.nlogs
        if not nlogs:
            nlogs = 50

        # Submit query
        params = {
            'type': 'log',
            'log-type': 'threat',
            'query': query,
            'nlogs': nlogs
        }

        response = conn.get(params, headers=headers)
        result = xmltodict.parse(response.data)
        job_id = result['response']['result']['job']

        # Poll until the job completes and fetch its entries
        entries = self._wait_and_fetch(conn, job_id, headers)

        self.response.payload = {
            'threats': [
                {
                    'threat_id': e['threatid'],
                    'type': e['subtype'],
                    'action': e['action'],
                    'src': e['src'],
                    'dst': e['dst'],
                    'app': e['app'],
                    'time': e['receive_time']
                }
                for e in entries
            ]
        }

    def _wait_and_fetch(self, conn, job_id, headers):
        import time

        for _ in range(40):
            time.sleep(3)
            params = {'type': 'log', 'action': 'get', 'job-id': job_id}
            response = conn.get(params, headers=headers)
            result = xmltodict.parse(response.data)

            if result['response']['result']['job']['status'] == 'FIN':
                entries = result['response']['result']['log']['logs'].get('entry', [])
                if not isinstance(entries, list):
                    entries = [entries] if entries else []
                return entries

        return []
```

## Query URL logs {#query-url-logs}

URL logs show access patterns and policy violations:

```python
# -*- coding: utf-8 -*-

from zato.server.service import Service

class GetUrlLogs(Service):
    name = 'security.paloalto.url-logs'

    input = '-query', '-nlogs'

    def handle(self):

        conn = self.rest['Panorama']
        headers = {'Authorization': 'Basic YOUR_BASE64_CREDENTIALS'}

        query = self.request.input.query

        nlogs = self.request.input.nlogs
        if not nlogs:
            nlogs = 100

        # Fetch URL logs using the pattern above
        entries = self._query_logs(conn, 'url', query, nlogs, headers)

        self.response.payload = {
            'url_logs': [
                {
                    'url': e['misc'],
                    'category': e['url_category_list'],
                    'action': e['action'],
                    'src': e['src'],
                    'dst': e['dst'],
                    'app': e['app'],
                    'time': e['receive_time']
                }
                for e in entries
            ]
        }

    def _query_logs(self, conn, log_type, query, nlogs, headers):
        import time
        import xmltodict

        params = {
            'type': 'log',
            'log-type': log_type,
            'query': query,
            'nlogs': nlogs
        }

        response = conn.get(params, headers=headers)
        result = xmltodict.parse(response.data)
        job_id = result['response']['result']['job']

        for _ in range(40):
            time.sleep(3)
            status_params = {'type': 'log', 'action': 'get', 'job-id': job_id}
            response = conn.get(status_params, headers=headers)
            result = xmltodict.parse(response.data)

            if result['response']['result']['job']['status'] == 'FIN':
                entries = result['response']['result']['log']['logs'].get('entry', [])
                if not isinstance(entries, list):
                    entries = [entries] if entries else []
                return entries

        return []
```

## Log entry mappers {#log-entry-mappers}

Every log type comes with its own fields - a mapper per type keeps the services above short:

```python
# -*- coding: utf-8 -*-

class TrafficLogMapper:
    def map(self, entry):
        return {
            'action': entry['action'],
            'app': entry['app'],
            'src': entry['src'],
            'dst': entry['dst'],
            'port': entry['dport'],
            'from_zone': entry['from'],
            'to_zone': entry['to'],
            'decrypted': entry['flag-proxy'],
            'device': entry['device_name'],
            'time': entry['receive_time']
        }


class ThreatLogMapper:
    def map(self, entry):
        return {
            'threat_id': entry['threatid'],
            'type': entry['subtype'],
            'action': entry['action'],
            'src': entry['src'],
            'dst': entry['dst'],
            'port': entry['dport'],
            'app': entry['app'],
            'device': entry['device_name'],
            'time': entry['receive_time']
        }


class DecryptionLogMapper:
    def map(self, entry):
        return {
            'cn': entry.get('cn'),
            'sni': entry.get('sni'),
            'error': entry['error'],
            'src': entry['src'],
            'dst': entry['dst'],
            'port': entry['dport'],
            'app': entry['app'],
            'device': entry['device_name'],
            'time': entry['receive_time']
        }
```

## See also {#see-also}

- [REST outgoing connections](https://zato.io/docs/dev/rest/outconns.html) - The connection type the Panorama examples use
- [NetBox](https://zato.io/docs/dev/examples/netbox.html) - Query and update the network source of truth
- [Scheduler](https://zato.io/docs/dev/examples/scheduler.html) - Run log queries on an interval

## Learn more {#learn-more}

- [Development documentation](https://zato.io/docs/dev/) - The complete reference for writing services
- [Requests and responses](https://zato.io/docs/dev/request-response/) - What a service receives, what it returns and how both are structured
- [Integration examples](https://zato.io/docs/dev/examples/) - Ready-made code for the systems you are likely to connect to
- [IDE and debugging](https://zato.io/docs/dev/ide/) - Write services in the Dashboard or in your own editor
- [Data models](https://zato.io/docs/dev/model/) - Declare inputs and outputs and have them validated for you
- [In-depth API tutorial](https://zato.io/tutorials/main/01.html) - The full platform tutorial, from installation to production patterns