Just built a Python script to alert on anomalous ZPA access patterns.

Having deployed Zscaler Private Access across our multi-cloud footprint, I've found its native reporting and alerting to be adequate for compliance but fundamentally reactive and lacking in predictive or anomaly-detection capabilities. The platform tells you *what* happened, but not when something is statistically unusual *as it's happening*. Given ZPA's role as a critical security boundary, I believe we need a layer of proactive monitoring on top of its logs.

To that end, I've developed a Python script that consumes ZPA's Audit Logs API, applies some basic statistical analysis, and pushes alerts to our observability stack (in this case, Datadog). The core idea is to establish a rolling baseline for "normal" access patterns and flag deviations that could indicate credential compromise, insider threat, or misconfigured application segments.

The script focuses on two key anomaly vectors:
* **Access Frequency Anomalies:** Sudden spikes in connection attempts from a single user or client IP, especially outside of that entity's established temporal pattern.
* **Geolocation Anomalies:** Successful authentications from geographic locations not previously observed for a given user, using a simple IP-to-geo lookup.

Here is the core anomaly detection logic. It uses a simple moving average and standard deviation model for frequency, which is lightweight enough to run on a scheduled Lambda or container.

```python
import pandas as pd
from datetime import datetime, timedelta
import requests
import geoip2.database

def detect_frequency_anomalies(logs_df, window_hours=24, sensitivity=2.5):
"""
logs_df: DataFrame with 'timestamp' and 'user' columns.
Returns DataFrame rows flagged as anomalous.
"""
logs_df['timestamp'] = pd.to_datetime(logs_df['timestamp'])
logs_df.set_index('timestamp', inplace=True)

anomalies = []
for user, group in logs_df.groupby('user'):
# Resample to hourly counts
hourly_counts = group.resample('H').size().rolling(window=window_hours, min_periods=1).mean()
hourly_std = group.resample('H').size().rolling(window=window_hours, min_periods=1).std().fillna(0)

current_count = group.last('H').size
baseline = hourly_counts.iloc[-1]
std = hourly_std.iloc[-1]

if current_count > (baseline + (sensitivity * std)):
anomalous_log = group.last('H').iloc[-1]
anomalies.append({
'user': user,
'observed_count': current_count,
'expected_baseline': baseline,
'standard_deviation': std,
'timestamp': anomalous_log.name
})
return pd.DataFrame(anomalies)

def check_geo_anomaly(user_ip, user_email, geo_reader, historical_map):
"""
geo_reader: geoip2 database reader object.
historical_map: dict mapping users to set of known country codes.
"""
try:
response = geo_reader.city(user_ip)
current_country = response.country.iso_code
except:
return None

known_countries = historical_map.get(user_email, set())
if known_countries and current_country not in known_countries:
return current_country
else:
# Update historical map
historical_map.setdefault(user_email, set()).add(current_country)
return None
```

The script then formats these anomalies as structured events and sends them via the Datadog API. The same pattern could easily adapt for Splunk, PagerDuty, or Slack webhooks.

**Current Limitations & Next Iterations:**
* The statistical model is rudimentary; I plan to replace it with an isolation forest or MAD (Median Absolute Deviation) for better outlier resistance.
* It currently processes logs in batch; a streaming implementation using Kinesis or Kafka would reduce detection latency.
* It lacks context from HR systems (e.g., is the user on vacation?) which would reduce false positives.
* The geo-database requires regular updates.

This is a starting point. The value isn't in the script itself, but in the principle: treat your ZPA logs as a first-class telemetry stream. The native dashboard is for hindsight; this layer is for insight. I'm interested if others in the community have built similar tooling or have ideas for more sophisticated anomaly vectors—perhaps around application sequencing or concurrent session analysis.

-- alex