Querying Palo Alto Strata Logging Service (SLS) Traffic and Threat Logs with Python
I was recently told that querying Palo Alto Networks Strata Logging Service (SLS) directly through an API was not possible.
I also found a few Reddit threads and older examples that pointed toward the legacy Cortex Data Lake APIs, but after spending some time with the Strata Cloud Manager Log Viewer and Chrome Developer Tools, it turns out that raw Traffic and Threat logs can be queried programmatically.
The important caveat is that I have not found public Palo Alto documentation for the specific Log Viewer resources used below. The API is clearly being used by the Strata Cloud Manager UI, though, and it works with the normal service-account OAuth flow.
API Endpoints
For the US tenant I tested against, Traffic logs were available here:
https://pa-us01.api.prismaaccess.com/api/sase/v3.0/resource/query/logviewer/firewall_trafficThreat logs use a similar resource:
https://pa-us01.api.prismaaccess.com/api/sase/v3.0/resource/query/logviewer/firewall_threatYour hostname may be different depending on region or tenant. If the URL above does not work for you, the easiest way to find the right one is to use Chrome Developer Tools while running a search in Strata Cloud Manager Log Viewer.
Open the Network tab and search for
firewall_traffic. You should be able to see the exact API
endpoint and the JSON payload the UI is sending.
Authentication
Authentication uses Palo Alto's documented OAuth 2.0 service-account flow. You need:
- Client ID
- Client Secret
- TSG ID
I used python-decouple to keep those values out of the source
code.
Install the dependencies:
pip install requests python-decouple
Then create a .env file:
PAN_CLIENT_ID=your_client_id
PAN_CLIENT_SECRET=your_client_secret
PAN_TSG_ID=your_tsg_idMake sure the file is excluded from source control:
echo ".env" >> .gitignoreOne Path That Did Not Work
I initially tried using the older Cortex Data Lake Query Service:
https://api.us.cdl.paloaltonetworks.com/query/v2/jobsUsing a modern SCM service-account JWT against that endpoint returned:
401
Jwt issuer is not configuredThat ended up being a useful clue. The modern Strata Cloud Manager service-account token and the older CDL Query Service authentication model are not interchangeable.
Once I switched over to the API endpoint that the SCM Log Viewer itself was using, authentication worked normally.
What the Query Looks Like
The Log Viewer API uses a JSON filter structure. A destination-IP search looks something like this:
{
"filter": {
"rules": [
{
"property": "event_time",
"operator": "between",
"values": [
1788460931984000,
1788461231984000
]
},
{
"property": "dest_ip.value",
"operator": "ip_match_cidr",
"values": [
"10.106.6.110"
]
}
]
},
"orderBy": {
"properties": [
{
"property": "time_generated",
"sort": {
"sequence": 1,
"order": "desc"
}
}
]
},
"pagination": {
"page_size": 100,
"page_number": 1
},
"count": 100
}The timestamps are Unix epoch values in microseconds.
Source and destination IPs are separate properties:
source_ip.value
dest_ip.value
The ip_match_cidr operator also makes it possible to search for
either a host or a subnet.
Python Example
The example below queries both Traffic and Threat logs, normalizes them into a common format, merges the results, and sorts everything newest-first.
It supports:
--source- source IP or CIDR--dest- destination IP or CIDR--port- destination port--application- Palo Alto App-ID--action- allow, deny, drop, reset, etc.--minutes- lookback window--max-logs- maximum combined results
import argparse
import ipaddress
import time
from datetime import datetime, timedelta, timezone
import requests
from decouple import config
CLIENT_ID = config("PAN_CLIENT_ID")
CLIENT_SECRET = config("PAN_CLIENT_SECRET")
TSG_ID = config("PAN_TSG_ID")
AUTH_URL = "https://auth.apps.paloaltonetworks.com/oauth2/access_token"
BASE_URL = (
"https://pa-us01.api.prismaaccess.com"
"/api/sase/v3.0/resource/query/logviewer"
)
TRAFFIC_URL = f"{BASE_URL}/firewall_traffic"
THREAT_URL = f"{BASE_URL}/firewall_threat"
class PrismaSASELogs:
def __init__(self):
self.token = None
self.token_expires_at = 0
self.session = requests.Session()
def authenticate(self):
response = requests.post(
AUTH_URL,
data={
"grant_type": "client_credentials",
"scope": f"tsg_id:{TSG_ID}",
},
auth=(CLIENT_ID, CLIENT_SECRET),
headers={
"Content-Type": "application/x-www-form-urlencoded",
},
timeout=30,
)
response.raise_for_status()
auth_data = response.json()
self.token = auth_data["access_token"]
expires_in = auth_data.get("expires_in", 899)
self.token_expires_at = (
time.time() + expires_in - 60
)
self.session.headers.update({
"Authorization": f"Bearer {self.token}",
"Prisma-Tenant": TSG_ID,
"Content-Type": "application/json",
"Accept": "application/json",
})
print("Successfully authenticated to Palo Alto.")
def ensure_authenticated(self):
if (
not self.token
or time.time() >= self.token_expires_at
):
self.authenticate()
@staticmethod
def datetime_to_microseconds(dt):
return int(dt.timestamp() * 1_000_000)
@staticmethod
def epoch_microseconds_to_datetime(value):
if not value:
return ""
dt = datetime.fromtimestamp(
value / 1_000_000,
tz=timezone.utc,
)
return dt.astimezone().strftime(
"%Y-%m-%d %H:%M:%S"
)
@staticmethod
def json_value(value):
if isinstance(value, dict):
return value.get("value", "")
return value if value is not None else ""
@staticmethod
def normalize_network(value):
if not value or value == "*":
return None
try:
network = ipaddress.ip_network(
value,
strict=False,
)
return str(network)
except ValueError as exc:
raise ValueError(
f"Invalid IP/subnet: {value}"
) from exc
def build_rules(
self,
start_time,
end_time,
source=None,
dest=None,
port=None,
action=None,
application=None,
):
rules = [
{
"property": "event_time",
"operator": "between",
"values": [
self.datetime_to_microseconds(start_time),
self.datetime_to_microseconds(end_time),
],
}
]
source = self.normalize_network(source)
dest = self.normalize_network(dest)
if source:
rules.append({
"property": "source_ip.value",
"operator": "ip_match_cidr",
"values": [source],
})
if dest:
rules.append({
"property": "dest_ip.value",
"operator": "ip_match_cidr",
"values": [dest],
})
if port is not None:
rules.append({
"property": "dest_port",
"operator": "eq",
"values": [int(port)],
})
if application and application != "*":
rules.append({
"property": "app",
"operator": "eq",
"values": [application],
})
if action and action.lower() != "any":
rules.append({
"property": "action.value",
"operator": "eq",
"values": [action.lower()],
})
return rules
def query_resource(
self,
url,
rules,
max_logs,
):
self.ensure_authenticated()
payload = {
"filter": {
"rules": rules,
},
"orderBy": {
"properties": [
{
"property": "time_generated",
"sort": {
"sequence": 1,
"order": "desc",
},
}
]
},
"pagination": {
"page_size": max_logs,
"page_number": 1,
},
"count": max_logs,
}
response = self.session.post(
url,
json=payload,
timeout=60,
)
if not response.ok:
print(f"\nQuery failed: {url}")
print("Status:", response.status_code)
print(response.text)
response.raise_for_status()
return response.json()
def normalize_common(self, log, log_type):
return {
"device": log.get(
"log_source_name",
"",
),
"log_type": log_type,
"time": self.epoch_microseconds_to_datetime(
log.get("time_generated")
),
"time_raw": log.get(
"time_generated",
0,
),
"source_ip": (
log.get(
"source_ip",
{},
).get(
"value",
"",
)
),
"source_port": log.get(
"source_port",
"",
),
"dest_ip": (
log.get(
"dest_ip",
{},
).get(
"value",
"",
)
),
"dest_port": log.get(
"dest_port",
"",
),
"application": log.get(
"app",
"",
),
"action": self.json_value(
log.get("action")
),
"from_zone": log.get(
"from_zone",
"",
),
"to_zone": log.get(
"to_zone",
"",
),
"rule": log.get(
"rule_matched",
"",
),
"source_user": log.get(
"source_user",
"",
),
"session_id": log.get(
"session_id"
),
"sequence_no": log.get(
"sequence_no"
),
"raw": log,
}
def normalize_traffic(self, log):
row = self.normalize_common(
log,
log_type="TRAFFIC",
)
row.update({
"session_end_reason": self.json_value(
log.get("session_end_reason")
),
"protocol": self.json_value(
log.get("protocol")
),
"bytes": log.get(
"bytes_total",
0,
),
"threat_name": "",
"threat_id": "",
})
return row
def normalize_threat(self, log):
row = self.normalize_common(
log,
log_type="THREAT",
)
row.update({
"session_end_reason": "",
"protocol": self.json_value(
log.get("protocol")
),
"bytes": 0,
"threat_name": (
log.get("threat_name")
or log.get("name")
or ""
),
"threat_id": (
log.get("threat_id")
or log.get("threatid")
or ""
),
})
return row
def search(
self,
minutes=15,
source=None,
dest=None,
port=None,
action=None,
application=None,
max_logs=100,
):
end_time = datetime.now(timezone.utc)
start_time = (
end_time
- timedelta(minutes=minutes)
)
rules = self.build_rules(
start_time=start_time,
end_time=end_time,
source=source,
dest=dest,
port=port,
action=action,
application=application,
)
print(
f"Searching last {minutes} minutes..."
)
traffic_response = self.query_resource(
TRAFFIC_URL,
rules,
max_logs,
)
threat_response = self.query_resource(
THREAT_URL,
rules,
max_logs,
)
traffic_logs = [
self.normalize_traffic(log)
for log in traffic_response.get(
"data",
[],
)
]
threat_logs = [
self.normalize_threat(log)
for log in threat_response.get(
"data",
[],
)
]
combined = (
traffic_logs
+ threat_logs
)
unique_logs = {}
for log in combined:
if log.get("sequence_no") is not None:
key = (
log["log_type"],
log["sequence_no"],
)
else:
key = (
log["log_type"],
log.get("session_id"),
log.get("time_raw"),
log.get("source_ip"),
log.get("source_port"),
log.get("dest_ip"),
log.get("dest_port"),
)
unique_logs[key] = log
results = list(
unique_logs.values()
)
results.sort(
key=lambda x: x["time_raw"],
reverse=True,
)
return results[:max_logs]
def print_logs(logs):
if not logs:
print("\nNo matching logs found.")
return
print()
print(
f"{'DEVICE':22} "
f"{'TYPE':8} "
f"{'TIME':19} "
f"{'SOURCE':22} "
f"{'DESTINATION':22} "
f"{'APP':18} "
f"{'ACTION':10} "
f"{'RULE / THREAT'}"
)
print("-" * 180)
for log in logs:
source = (
f"{log['source_ip']}:"
f"{log['source_port']}"
)
destination = (
f"{log['dest_ip']}:"
f"{log['dest_port']}"
)
if log["log_type"] == "THREAT":
detail = log["threat_name"]
else:
detail = log["rule"]
print(
f"{log['device']:<22.22} "
f"{log['log_type']:<8} "
f"{log['time']:<19} "
f"{source:<22.22} "
f"{destination:<22.22} "
f"{log['application']:<18.18} "
f"{log['action']:<10.10} "
f"{detail}"
)
print()
print(
f"{len(logs)} log entries returned."
)
def main():
parser = argparse.ArgumentParser(
description=(
"Search Prisma Access traffic and threat logs."
)
)
parser.add_argument(
"--minutes",
type=int,
default=15,
help=(
"Search window in minutes. "
"Default: 15"
),
)
parser.add_argument(
"--source",
default="*",
help=(
"Source IP or subnet. "
"Default: *"
),
)
parser.add_argument(
"--dest",
default="*",
help=(
"Destination IP or subnet. "
"Default: *"
),
)
parser.add_argument(
"--port",
type=int,
default=None,
help="Destination port",
)
parser.add_argument(
"--action",
choices=[
"any",
"allow",
"deny",
"drop",
"reset-client",
"reset-server",
"reset-both",
],
default="any",
help=(
"Traffic/threat action. "
"Default: any"
),
)
parser.add_argument(
"--application",
default="*",
help=(
"Application name. "
"Default: *"
),
)
parser.add_argument(
"--max-logs",
type=int,
default=100,
help=(
"Maximum combined traffic/threat "
"results. Default: 100"
),
)
args = parser.parse_args()
if args.minutes <= 0:
parser.error(
"--minutes must be greater than zero"
)
if not 1 <= args.max_logs <= 1000:
parser.error(
"--max-logs must be between 1 and 1000"
)
if args.port is not None:
if not 1 <= args.port <= 65535:
parser.error(
"--port must be between 1 and 65535"
)
try:
prisma = PrismaSASELogs()
logs = prisma.search(
minutes=args.minutes,
source=args.source,
dest=args.dest,
port=args.port,
action=args.action,
application=args.application,
max_logs=args.max_logs,
)
print_logs(logs)
except ValueError as exc:
parser.error(str(exc))
except requests.HTTPError as exc:
print(
f"\nHTTP error: {exc}"
)
raise
except requests.RequestException as exc:
print(
f"\nNetwork/API error: {exc}"
)
raise
if __name__ == "__main__":
main()
Running It
A simple destination-IP search:
python3.11 run.py --dest 10.106.6.110 --minutes 30A more specific query:
python3.11 run.py \
--dest 10.106.6.110 \
--port 8080 \
--application web-browsing \
--minutes 30You can also search by source:
python3.11 run.py \
--source 10.255.10.201 \
--minutes 60
Use --help to see all of the available arguments:
python3.11 run.py --helpExample Output
Final Thoughts
I would still consider this an undocumented integration point rather than a formally supported public API contract. Palo Alto could change the resource names or query format in the future.
That said, if your goal is to programmatically search Prisma Access Traffic and Threat logs, this works today and is much cleaner than trying to scrape the Log Viewer UI with Selenium.
The biggest lesson for me was that the browser developer tools were more useful than the public documentation for finding the actual Log Viewer API. Once the endpoint was identified, the rest was straightforward:
- Authenticate with a normal Strata Cloud Manager service account.
- Use the regional Prisma Access API hostname.
- POST the same filter structure used by the Log Viewer UI.
- Parse the returned
datarecords.
Hopefully this saves somebody else from spending a few hours going down the older Cortex Data Lake API path.
Comments
Post a Comment