Panther high experimental python

Okta SWA Bulk Access, New Source, and Credential Extraction - Behavioral

Detects Okta SWA (Secure Web Authentication) bulk credential extraction, abuse, and access from previously unseen IP addresses or user agents using behavioral z-score and source novelty analysis. SWA apps store credentials in Okta's encrypted vault. Admin accounts with SWA access can view or rotate credentials for users across many apps. This detection builds a 90-day behavioral baseline for each admin's SWA access, credential change patterns, and known source IPs/user agents, then identifies anomalous spikes or new sources in the last 7 days. **Detection Logic:** - Z-score: SWA authentication volume spike (> 3σ above baseline) - Z-score: Unique SWA app diversity spike (many different apps accessed in one hour) (> 3σ) - Z-score: Credential extraction volume spike (> 3σ) - Z-score: Victim diversity spike (credential changes across many users) (> 2σ) - Cold-start: First-time bulk SWA access (>= 10 events, no prior baseline) - Cold-start: First-time credential extraction (>= 5 extractions, no prior baseline) - New source: SWA access from IP address not seen in 90-day baseline - New source: SWA access from user agent not seen in 90-day baseline - Critical compound: New IP + any credential extraction events **Why This Matters:** SWA credential extraction is a powerful lateral movement technique. An attacker with admin access can silently retrieve plaintext credentials for hundreds of SWA-protected applications without triggering MFA or generating obvious authentication failures. New source detection catches the initial access phase when a compromised admin account is used from an unfamiliar device or location. **Complementary Detection:** Use alongside `Okta.SWA.OffHoursAccess.Behavioral` which detects the same attack vector occurring outside normal business hours.

View Source

Detection Logic

def rule(event):
    # Query already filtered for is_anomalous = TRUE.
    # Guard against malformed rows missing the primary key field.
    return bool(event.get("admin_email"))


def title(event):
    admin = event.get("admin_email", "Unknown")
    recent_extractions = event.get("recent_total_extractions") or 0
    recent_swa_events = event.get("recent_total_swa_events") or 0
    has_new_ip = event.get("has_new_ip") or False
    new_ip_extraction_count = event.get("new_ip_extraction_count") or 0
    if recent_extractions > 0 and has_new_ip:
        return (
            f"Okta SWA: Credential Extraction from New IP by {admin}"
            f" ({new_ip_extraction_count} extractions from new source)"
        )
    if recent_extractions > 0:
        return f"Okta SWA: Bulk Credential Extraction by {admin} ({recent_extractions} extractions)"
    if has_new_ip:
        return f"Okta SWA: Access from New IP by {admin} ({recent_swa_events} events)"
    return f"Okta SWA: Bulk App Access Anomaly by {admin} ({recent_swa_events} events)"


def severity(event):
    is_first_extraction = event.get("is_first_time_credential_extraction") or False
    is_extraction_anomaly = event.get("is_extraction_volume_anomaly") or False
    is_victim_anomaly = event.get("is_victim_diversity_anomaly") or False
    recent_extractions = event.get("recent_total_extractions") or 0
    has_new_ip = event.get("has_new_ip") or False
    has_new_user_agent = event.get("has_new_user_agent") or False
    new_ip_extraction_count = event.get("new_ip_extraction_count") or 0
    score = event.get("anomaly_severity_score") or 0
    if has_new_ip and new_ip_extraction_count > 0:
        return "CRITICAL"
    if is_first_extraction or is_extraction_anomaly or is_victim_anomaly:
        return "CRITICAL"
    if has_new_ip or recent_extractions > 0 or score > 20:
        return "HIGH"
    if has_new_user_agent:
        return "MEDIUM"
    return "MEDIUM"  # Default: anomalous SWA volume with no other escalating signals


def dedup_key(event):
    admin = event.get("admin_email", "unknown")
    first_event = str(
        event.get("recent_extraction_first_event") or event.get("recent_swa_first_event", "unknown")
    )[:10]
    return f"okta_swa_bulk_{admin}_{first_event}"


def alert_context(event):
    return {
        "admin_email": event.get("admin_email"),
        "recent_total_extractions": event.get("recent_total_extractions"),
        "recent_max_victim_diversity_per_hour": event.get("recent_max_victim_diversity_per_hour"),
        "recent_total_swa_events": event.get("recent_total_swa_events"),
        "recent_max_app_diversity_per_hour": event.get("recent_max_app_diversity_per_hour"),
        "z_score_extraction_volume": event.get("z_score_extraction_volume"),
        "z_score_victim_diversity": event.get("z_score_victim_diversity"),
        "z_score_swa_volume": event.get("z_score_swa_volume"),
        "has_new_ip": event.get("has_new_ip"),
        "has_new_user_agent": event.get("has_new_user_agent"),
        "new_ip_count": event.get("new_ip_count"),
        "new_ip_extraction_count": event.get("new_ip_extraction_count"),
        "new_ip_victim_count": event.get("new_ip_victim_count"),
        "anomaly_severity_score": event.get("anomaly_severity_score"),
        "is_first_time_credential_extraction": event.get("is_first_time_credential_extraction"),
        "recent_extraction_first_event": event.get("recent_extraction_first_event"),
        "recent_extraction_last_event": event.get("recent_extraction_last_event"),
    }

Field Validations

Loading…

Comments (0)

Loading comments...