Panther rules: okta

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


def title(event):
    user_email = event.get("user_email", "<UNKNOWN_USER>")
    severity_score = event.get("anomaly_severity_score", 0)

    return (
        f"Okta AD Agent Authentication Anomaly Detected: {user_email} "
        f"(Severity Score: {severity_score})"
    )


def severity(event):
    # Dynamic severity based on anomaly severity score and z-score magnitudes.
    # Higher z-scores = more standard deviations from baseline = more suspicious.
    # Cold-start events have null z-scores (no baseline), so default to 0.
    severity_score = event.get("anomaly_severity_score") or 0
    z_volume = event.get("z_score_volume") or 0
    z_ip = event.get("z_score_ip_diversity") or 0
    z_country = event.get("z_score_country_diversity") or 0

    # Critical: Extreme anomaly (severity score > 15 or any z-score > 5)
    if severity_score > 15 or max(z_volume, z_ip, z_country) > 5:
        return "CRITICAL"

    # High: Strong anomaly (severity score > 10 or any z-score > 4)
    if severity_score > 10 or max(z_volume, z_ip, z_country) > 4:
        return "HIGH"

    # Medium: Moderate anomaly (default for detections that passed threshold)
    return "MEDIUM"


def alert_context(event):
    return {
        # User information
        "user_email": event.get("user_email", "<UNKNOWN_USER>"),
        # Baseline behavior
        "baseline_total_events": event.get("baseline_total_events", 0),
        "baseline_active_days": event.get("baseline_active_days", 0),
        "baseline_mean_events_per_hour": event.get("baseline_mean_events_per_hour", 0),
        "baseline_mean_ip_diversity": event.get("baseline_mean_ip_diversity_per_hour", 0),
        "baseline_mean_country_diversity": event.get("baseline_mean_country_diversity_per_hour", 0),
        # Recent anomalous activity
        "recent_total_events": event.get("recent_total_events", 0),
        "recent_max_events_per_hour": event.get("recent_max_events_per_hour", 0),
        "recent_max_ip_diversity": event.get("recent_max_ip_diversity_per_hour", 0),
        "recent_max_country_diversity": event.get("recent_max_country_diversity_per_hour", 0),
        "recent_max_device_diversity": event.get("recent_max_device_diversity_per_hour", 0),
        # Z-scores (standard deviations from baseline)
        "z_score_volume": event.get("z_score_volume", 0),
        "z_score_ip_diversity": event.get("z_score_ip_diversity", 0),
        "z_score_country_diversity": event.get("z_score_country_diversity", 0),
        "z_score_device_diversity": event.get("z_score_device_diversity", 0),
        "anomaly_severity_score": event.get("anomaly_severity_score", 0),
        # Geographic and network context
        "recent_ip_addresses": event.get("all_recent_ips", []),
        "recent_countries": event.get("all_recent_countries", []),
        # Temporal context
        "first_anomaly_hour": event.get("first_anomaly_hour", "<UNKNOWN>"),
        "last_anomaly_hour": event.get("last_anomaly_hour", "<UNKNOWN>"),
        "detection_timestamp": event.get("detection_timestamp", "<UNKNOWN>"),
    }


def dedup_key(event):
    # Deduplicate by user and hour to avoid alert spam during active attacks.
    user = event.get("user_email", "unknown")
    first_hour = str(event.get("first_anomaly_hour", "unknown"))

    return f"okta_ad_agent_zscore_anomaly_{user}_{first_hour}"

AnalysisType: scheduled_rule
Filename: okta_ad_agent_auth_zscore_anomaly.py
RuleID: "Okta.ADAgent.AuthenticationAnomaly.ZScore"
DisplayName: "Okta AD Agent Authentication Anomaly - Z-Score Detection"
Enabled: false  # Start disabled for tuning
ScheduledQueries:
  - Query.Okta.ADAgentAuthZScoreAnomaly
Severity: Medium  # Default, dynamic severity in rule function
Status: Experimental
Tags:
  - Identity & Access Management
  - Okta
  - Active Directory
  - Credential Access:Steal Application Access Token
  - Credential Access:Brute Force
  - Initial Access:Valid Accounts
  - Anomaly Detection
  - Statistical Analysis
Reports:
  MITRE ATT&CK:
    - TA0006:T1528  # Steal Application Access Token
    - TA0006:T1110  # Brute Force
    - TA0001:T1078  # Valid Accounts
Description: |
  Detects potential Okta AD Agent token theft and credential abuse using statistical z-score analysis.

  This detection uses a lookup table containing 90-day behavioral baselines for each user's AD Agent
  authentication patterns, then calculates z-scores to identify suspicious activity in the last 7 days.

  **PREREQUISITES:**
  1. Baseline builder query must run first: `Query.Okta.ADAgentBaselineBuilder`
  2. Lookup table must be configured: `okta_ad_pantherflow_baseline_90d`
  3. Allow 24 hours for initial baseline to populate

  **Detection Logic:**
  - Calculates mean and standard deviation for hourly authentication volume, IP diversity,
    country diversity, and device diversity
  - Alerts when recent activity shows BOTH:
    1. Volume spike (z-score > 3 standard deviations)
    2. Geographic/IP diversity spike (z-score > 2 standard deviations)

  **Why This Matters:**
  Token theft attacks have a distinct signature: stolen credentials are used from multiple
  locations/IPs simultaneously or in rapid succession. This creates both a volume spike and
  a diversity spike that this detection identifies.

  **Complementary Detection:**
  This rule complements `Okta.ADAgent.TokenAbuse.Behavioral` which detects admin actions
  (token creation, agent configuration) from new sources. This rule detects the actual USE
  of stolen tokens through authentication patterns.

Reference: https://www.varonis.com/blog/okta-attack-vectors
Runbook: |
  1. Compare recent_max_events_per_hour against baseline_mean_events_per_hour for user_email and review z_score_volume and z_score_ip_diversity to quantify the anomaly - confirm baseline_updated_at is within the past 7 days to ensure the baseline reflects current normal behavior
  2. Check all_recent_ips and all_recent_countries for geographic anomalies in the 7 days around first_anomaly_hour - look for simultaneous access from multiple countries or rapid location changes inconsistent with the user's primary_country and primary_ip
  3. Search Okta SystemLog for system.api_token.create and system.agent.ad.agent_instance_added events by user_email in the 48 hours before first_anomaly_hour, and check for Okta.ADAgent.TokenAbuse.Behavioral alerts from this user in the past 7 days

DedupPeriodMinutes: 360  # 6 hours, match query frequency
SummaryAttributes:
  - user_email
  - anomaly_severity_score
  - z_score_volume

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


def title(event):
    actor = event.get("actorId", "<UNKNOWN_ACTOR>")
    event_type = event.get("eventType", "<UNKNOWN_EVENT>")
    anomaly_type = event.get("anomaly_type", "Unknown Anomaly")

    return f"Okta AD Agent Activity from {anomaly_type}: {actor} - {event_type}"


def severity(event):
    event_type = event.get("eventType", "")

    # New agent registration is critical (potential rogue agent)
    if "agent_instance_added" in event_type:
        return "CRITICAL"

    # Token creation from new source is high severity
    if "api_token.create" in event_type:
        return "HIGH"

    # Config changes are medium severity
    if "config_change" in event_type:
        return "MEDIUM"

    return "MEDIUM"


def dedup_key(event):
    actor = event.get("actorId", "unknown")
    event_date = str(event.get("p_event_time", "unknown"))[:10]
    return f"okta_ad_agent_token_abuse_{actor}_{event_date}"


def alert_context(event):
    return {
        "actor_id": event.get("actorId", "<UNKNOWN_ACTORID>"),
        "actor_name": event.get("actorName", "<UNKNOWN_ACTORNAME>"),
        "event_type": event.get("eventType", "<UNKNOWN_EVENTTYPE>"),
        "source_ip": event.get("sourceIP", "<UNKNOWN_SOURCEIP>"),
        "user_agent": event.get("userAgent", "<UNKNOWN_USERAGENT>"),
        "anomaly_type": event.get("anomaly_type", "<UNKNOWN_ANOMALYTYPE>"),
        "target": event.get("target", []),
        "event_time": event.get("p_event_time", "<UNKNOWN_EVENTTIME>"),
    }

AnalysisType: scheduled_rule
Filename: okta_ad_agent_token_abuse_behavioral.py
RuleID: "Okta.ADAgent.TokenAbuse.Behavioral"
DisplayName: "Okta AD Agent Token Abuse - Behavioral"
Enabled: true
ScheduledQueries:
  - Query.Okta.ADAgentTokenAbuseBehavioral
Severity: High
Status: Experimental
Tags:
  - Identity & Access Management
  - Okta
  - Active Directory
  - Credential Access:Steal Application Access Token
  - Persistence:Account Manipulation
  - Anomaly Detection
Reports:
  MITRE ATT&CK:
    - TA0006:T1528
    - TA0003:T1098
Description: |
  Detects potential Okta AD Agent token theft and abuse using behavioral analysis.
  Instead of relying on hardcoded service account patterns, this detection identifies
  when AD agent-related activities (API token creation, agent registration, config changes)
  occur from previously unseen IP addresses or user agents. This behavioral approach
  adapts to your environment and catches anomalous access patterns that may indicate
  compromised credentials or unauthorized token generation.

  **What This Detection Catches:**
  - API token creation from new IPs or user agents
  - New AD agent registrations from unexpected sources
  - AD agent configuration changes from new locations

  **Complementary Detection:**
  Use alongside `Okta.ADAgent.AuthenticationAnomaly.ZScore` which detects the actual
  USE of stolen tokens through authentication pattern anomalies.

Reference: https://www.varonis.com/blog/okta-attack-vectors
Runbook: |
  1. Query Okta SystemLog for all events by actorId in the 24 hours before and after the alert, focusing on system.api_token.create, system.agent.ad.agent_instance_added, and system.agent.ad.config_change_detected events to establish the full scope of activity
  2. Verify whether sourceIP and userAgent have been seen for actorId in the past 30 days, and check if sourceIP is associated with known corporate or administrative network ranges
  3. Search for other alerts from actorId or sourceIP in the past 7 days, including Okta.ADAgent.AuthenticationAnomaly.ZScore detections that may indicate a stolen token is actively being used

DedupPeriodMinutes: 1440  # 24 hours — matches dedup_key anchor (actor + event_date)
SummaryAttributes:
  - eventType
  - anomaly_type
  - actorId

AnalysisType: scheduled_query
QueryName: "Okta Admin Access Granted"
Enabled: false
Description: >
  Audit instances of admin access granted in your okta tenant
SnowflakeQuery: |
  SELECT
  p_event_time as event_time,
  actor:alternateId as actor_email,
  actor:displayName as actor_name,
  displayMessage,
  eventType,
  debugContext:debugData:privilegeGranted as priv_granted,
  target as target_name,
  client:ipAddress as src_ip,
  client:geographicalContext:city as city,
  client:geographicalContext:country as country,
  client:userAgent:rawUserAgent as user_agent
  FROM
    panther_logs.public.okta_systemlog
  WHERE
  ( eventType = 'user.account.privilege.grant'
   OR
    eventType = 'group.privilege.grant'
   AND
     debugContext:debugData:privilegeGranted like '%Admin%'
  )
    AND
    p_occurs_between('2022-01-14','2022-03-22')
  ORDER BY
  event_time desc

DatabricksQuery: |
  SELECT
  p_event_time as event_time,
  actor:alternateId as actor_email,
  actor:displayName as actor_name,
  displayMessage,
  eventType,
  debugContext:debugData:privilegeGranted as priv_granted,
  target as target_name,
  client:ipAddress as src_ip,
  client:geographicalContext:city as city,
  client:geographicalContext:country as country,
  client:userAgent:rawUserAgent as user_agent
  FROM
    panther_logs.okta_systemlog
  WHERE
  ( eventType = 'user.account.privilege.grant'
   OR
    eventType = 'group.privilege.grant'
   AND
     debugContext:debugData:privilegeGranted like '%Admin%'
  )
    AND
    p_occurs_between('2022-01-14','2022-03-22')
  ORDER BY
  event_time desc
Schedule:
  RateMinutes: 43200
  TimeoutMinutes: 1

import re

from panther_okta_helpers import okta_alert_context

ADMIN_PATTERN = re.compile(r"[aA]dministrator")


def rule(event):
    return (
        event.get("eventType", None) == "user.account.privilege.grant"
        and event.deep_get("outcome", "result") == "SUCCESS"
        and bool(
            ADMIN_PATTERN.search(
                event.deep_get("debugContext", "debugData", "privilegeGranted", default="")
            )
        )
    )


def dedup(event):
    return event.deep_get("debugContext", "debugData", "requestId", default="<UNKNOWN_REQUEST_ID>")


def title(event):
    target = event.get("target", [{}])
    display_name = target[0].get("displayName", "MISSING DISPLAY NAME") if target else ""
    alternate_id = target[0].get("alternateId", "MISSING ALTERNATE ID") if target else ""
    privilege = event.deep_get(
        "debugContext", "debugData", "privilegeGranted", default="<UNKNOWN_PRIVILEGE>"
    )

    return (
        f"{event.deep_get('actor', 'displayName')} "
        f"<{event.deep_get('actor', 'alternateId')}> granted "
        f"[{privilege}] privileges to {display_name} <{alternate_id}>"
    )


def alert_context(event):
    return okta_alert_context(event)


def severity(event):
    if "Super administrator" in event.deep_get(
        "debugContext", "debugData", "privilegeGranted", default=""
    ):
        return "HIGH"
    return "INFO"

AnalysisType: rule
Filename: okta_admin_role_assigned.py
RuleID: "Okta.AdminRoleAssigned"
DisplayName: "Okta Admin Role Assigned"
Enabled: true
LogTypes:
  - Okta.SystemLog
Tags:
  - Identity & Access Management
  - Okta
  - Privilege Escalation:Valid Accounts
Reports:
  MITRE ATT&CK:
    - TA0004:T1078
Severity: Info
Description: A user has been granted administrative privileges in Okta
Reference: https://help.okta.com/en/prod/Content/Topics/Security/administrators-admin-comparison.htm
Runbook: Reach out to the user if needed to validate the activity
DedupPeriodMinutes: 15
SummaryAttributes:
  - eventType
  - severity
  - displayMessage
  - p_any_ip_addresses

from panther_okta_helpers import okta_alert_context


def rule(event):
    return (
        event.get("eventType") == "user.authentication.auth_via_mfa"
        and event.deep_get("outcome", "result") == "FAILURE"
        and event.deep_get("outcome", "reason") == "FastPass declined phishing attempt"
    )


def title(event):
    return (
        f"{event.deep_get('actor', 'displayName', default='<displayName-not-found>')} "
        f"<{event.deep_get('actor', 'alternateId', default='alternateId-not-found')}> "
        f"FastPass declined phishing attempt"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_phishing_attempt_blocked_by_fastpass.py
RuleID: "Okta.Phishing.Attempt.Blocked.FastPass"
DisplayName: "Okta AiTM Phishing Attempt Blocked by FastPass"
Enabled: true
LogTypes:
  - Okta.SystemLog
Reports:
  MITRE ATT&CK:
    - TA0001:T1566 # Phishing
    - TA0006:T1556 # Modify Authentication Process
    - TA0003:T1078.004 # Valid Accounts: Cloud Accounts
Severity: High
Description: >
  Okta FastPass detected a user targeted by attackers wielding real-time (AiTM) proxies.
Runbook: >
  Protect sign-in flows by enforcing phishing-resistant authentication with Okta FastPass and FIDO2 WebAuthn.
Reference: >
  https://sec.okta.com/fastpassphishingdetection
DedupPeriodMinutes: 30
Threshold: 1

from panther_okta_helpers import okta_alert_context


def rule(event):
    return (
        event.get("eventType", None) == "system.api_token.create"
        and event.deep_get("outcome", "result") == "SUCCESS"
    )


def title(event):
    target = event.get("target", [{}])
    key_name = target[0].get("displayName", "MISSING DISPLAY NAME") if target else "MISSING TARGET"

    return (
        f"{event.deep_get('actor', 'displayName')} <{event.deep_get('actor', 'alternateId')}>"
        f"created a new API key - <{key_name}>"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_api_key_created.py
RuleID: "Okta.APIKeyCreated"
DisplayName: "Okta API Key Created"
Enabled: true
LogTypes:
  - Okta.SystemLog
Tags:
  - Identity & Access Management
  - Okta
  - Credential Access:Steal Application Access Token
Reports:
  MITRE ATT&CK:
    - TA0006:T1528
Severity: Info
Description: A user created an API Key in Okta
Reference: https://help.okta.com/en/prod/Content/Topics/Security/API.htm
Runbook: Reach out to the user if needed to validate the activity.
SummaryAttributes:
  - eventType
  - severity
  - displayMessage
  - p_any_ip_addresses

from panther_okta_helpers import okta_alert_context


def rule(event):
    return (
        event.get("eventType", None) == "system.api_token.revoke"
        and event.deep_get("outcome", "result") == "SUCCESS"
    )


def title(event):
    target = event.get("target", [{}])
    key_name = target[0].get("displayName", "MISSING DISPLAY NAME") if target else "MISSING TARGET"

    return (
        f"{event.deep_get('actor', 'displayName')} <{event.deep_get('actor', 'alternateId')}>"
        f"revoked API key - <{key_name}>"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_api_key_revoked.py
RuleID: "Okta.APIKeyRevoked"
DisplayName: "Okta API Key Revoked"
Enabled: true
LogTypes:
  - Okta.SystemLog
Tags:
  - Identity & Access Management
  - Okta
Severity: Info
Description: A user has revoked an API Key in Okta
Reference: https://help.okta.com/en/prod/Content/Topics/Security/API.htm
Runbook: Validate this action was authorized.
SummaryAttributes:
  - eventType
  - severity
  - displayMessage
  - p_any_ip_addresses

from panther_okta_helpers import okta_alert_context


def rule(event):
    return event.get("eventtype") in (
        "app.oauth2.as.token.detect_reuse",
        "app.oauth2.token.detect_reuse",
    )


def title(event):
    return (
        "Okta Access Token Reuse Attempted by "
        f"[{event.get('client', {}).get('ipAddress')}] "
        f"[{event.get('actor', {}).get('displayName', '<no-displayname-found>')}]"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Description: |-
  When a client wants to renew an access token, it sends the refresh token with the access token request to the /token Okta endpoint.
  Okta validates the incoming refresh token, issues a new set of tokens and invalidates the refresh token that was passed with the initial request.
  This detection alerts when a previously used refresh token is used again with the token request
Reference: https://developer.okta.com/docs/guides/refresh-tokens/main/#refresh-token-reuse-detection
DisplayName: "Okta App Refresh Access Token Reuse"
Enabled: true
Filename: okta_app_refresh_access_token_reuse.py
Runbook: Determine if the clientip is anomalous. Revoke tokens if deemed suspicious.
Severity: Medium
DedupPeriodMinutes: 60
LogTypes:
  - Okta.SystemLog
RuleID: "Okta.Refresh.Access.Token.Reuse"
Threshold: 1

from panther_okta_helpers import okta_alert_context


def rule(event):
    return event.get("eventtype", "") == "app.generic.unauth_app_access_attempt"


def title(event):
    return (
        f"[{event.deep_get('actor', 'alternateId', default = '<id-not-found>')}] "
        f"attempted unauthorized access to "
        f"[{event.get('target', [{}])[0].get('alternateId','<id-not-found>')}]"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Description: Detects when a user is denied access to an Okta application
DisplayName: "Okta App Unauthorized Access Attempt"
Enabled: true
Filename: okta_app_unauthorized_access_attempt.py
Severity: Low
Reference: https://support.okta.com/help/s/article/App-Sign-on-Error-403-User-attempted-unauthorized-access-to-app?language=en_US
DedupPeriodMinutes: 60
LogTypes:
  - Okta.SystemLog
RuleID: "Okta.App.Unauthorized.Access.Attempt"
Threshold: 5

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_weakenings = event.get("recent_total_weakenings") or 0
    recent_admin_enrollments = event.get("recent_total_admin_enrollments") or 0
    if recent_weakenings > 0:
        return f"Okta Skeleton Key: Security Policy Weakening by {admin}"
    if recent_admin_enrollments > 0:
        return f"Okta Skeleton Key: Bulk Admin Factor Enrollment by {admin}"
    return f"Okta Skeleton Key Bypass Anomaly Detected for {admin}"


def severity(event):
    score = event.get("anomaly_severity_score") or 0
    is_first_security_weakening = event.get("is_first_time_security_weakening") or False
    recent_weakenings = event.get("recent_total_weakenings") or 0
    if is_first_security_weakening or (recent_weakenings > 0 and score > 20):
        return "CRITICAL"
    if recent_weakenings > 0 or score > 15:
        return "HIGH"
    return "MEDIUM"


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


def alert_context(event):
    return {
        "admin_email": event.get("admin_email"),
        "recent_total_weakenings": event.get("recent_total_weakenings"),
        "recent_total_admin_enrollments": event.get("recent_total_admin_enrollments"),
        "z_score_security_weakenings": event.get("z_score_security_weakenings"),
        "z_score_admin_enrollments": event.get("z_score_admin_enrollments"),
        "anomaly_severity_score": event.get("anomaly_severity_score"),
        "is_first_time_security_weakening": event.get("is_first_time_security_weakening"),
        "is_first_time_admin_enrollment": event.get("is_first_time_admin_enrollment"),
        "baseline_total_weakenings": event.get("baseline_total_weakenings"),
        "recent_policy_first_event": event.get("recent_policy_first_event"),
        "recent_policy_last_event": event.get("recent_policy_last_event"),
    }

AnalysisType: scheduled_rule
Filename: okta_skeleton_key_bypass_behavioral.py
RuleID: "Okta.SkeletonKeyBypass.Behavioral"
DisplayName: "Okta Authentication Bypass via Skeleton Key Injection - Behavioral"
Enabled: true
ScheduledQueries:
  - Query.Okta.SkeletonKeyBypassBehavioral
Severity: High  # Default, dynamic severity in rule function
Status: Experimental
Tags:
  - Identity & Access Management
  - Okta
  - Active Directory
  - Defense Evasion:Modify Authentication Process
  - Persistence:Account Manipulation
  - Anomaly Detection
Reports:
  MITRE ATT&CK:
    - TA0005:T1556  # Modify Authentication Process
    - TA0003:T1098  # Account Manipulation
Description: |
  Detects potential Okta authentication bypass via skeleton key injection using behavioral z-score analysis.

  Skeleton key attacks in Okta involve manipulating authentication policies to weaken MFA requirements
  (disabling requireFactor, zeroing maxSessionLifetime) and bulk-enrolling attacker-controlled
  authenticators on victim accounts. This detection builds a 90-day behavioral baseline for each
  admin's policy change and factor enrollment patterns, then identifies anomalous spikes in the last
  7 days.

  **Detection Logic:**
  - Z-score: Spike in security-weakening policy changes (> 2σ above baseline)
  - Z-score: Spike in admin-on-behalf-of MFA factor enrollments (> 3σ above baseline)
  - Cold-start: First-time security weakening (no prior baseline - immediate high-confidence signal)
  - Cold-start: First-time admin-enrolled factors for other users

  **Why This Matters:**
  Skeleton key attacks require two steps: weaken authentication policies to reduce MFA friction,
  then enroll attacker-controlled authenticators on victim accounts. This detection catches both
  steps using behavioral baselines that adapt to legitimate admin workflows.

  **Complementary Detection:**
  Use alongside `Okta.ADAgent.TokenAbuse.Behavioral` for admin credential theft scenarios.

Reference: https://www.varonis.com/blog/okta-attack-vectors
Runbook: |
  1. Review recent_total_weakenings and recent_max_weakenings_per_hour for admin_email against baseline_total_weakenings - query Okta SystemLog for policy.rule.update and policy.lifecycle.update events by admin_email in the 24 hours around recent_policy_first_event, focusing on changedAttributes containing requireFactor=false or maxSessionLifetimeMinutes=0
  2. Check recent_total_admin_enrollments and z_score_admin_enrollments for user.mfa.factor.activate events by admin_email in the 7 days around recent_enrollment_first_event - identify target accounts enrolled and verify whether enrollment was authorized and factor types are consistent with corporate standards
  3. Search for Okta.ADAgent.TokenAbuse.Behavioral or other privileged account alerts for admin_email in the 48 hours before recent_policy_first_event to determine whether the admin account itself was compromised prior to the policy manipulation

DedupPeriodMinutes: 1440  # 24 hours
SummaryAttributes:
  - admin_email
  - anomaly_severity_score
  - recent_total_weakenings

from panther_okta_helpers import okta_alert_context


def rule(event):
    return event.get(
        "eventType"
    ) == "application.lifecycle.update" and "Pushing user passwords" in event.deep_get(
        "outcome", "reason", default=""
    )


def title(event):
    target = event.deep_walk(
        "target", "alternateId", default="<alternateId-not-found>", return_val="first"
    )
    return (
        f"{event.deep_get('actor', 'displayName', default='<displayName-not-found>')} "
        f"<{event.deep_get('actor', 'alternateId', default='alternateId-not-found')}> "
        f"extracted cleartext user passwords via SCIM app [{target}]"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_password_extraction_via_scim.py
RuleID: "Okta.Password.Extraction.via.SCIM"
DisplayName: "Okta Cleartext Passwords Extracted via SCIM Application"
Enabled: true
LogTypes:
  - Okta.SystemLog
Reports:
  MITRE ATT&CK:
    - TA0006:T1556 # Modify Authentication Process
Severity: High
Description: >
  An application admin has extracted cleartext user passwords via SCIM app.
  Malcious actors can extract plaintext passwords by creating a SCIM application under their control and configuring it to sync passwords from Okta.
Reference: >
  https://www.authomize.com/blog/authomize-discovers-password-stealing-and-impersonation-risks-to-in-okta/
DedupPeriodMinutes: 30
Threshold: 1

from panther_okta_helpers import okta_alert_context


def rule(event):
    return event.get("eventtype", "") == "group.privilege.grant"


def title(event):
    # pylint: disable=W0613
    return (
        "Okta Admin Privileges Assigned to Group "
        f"[{event.get('target', [{}])[0].get('alternateId', '<id-not-found>')}]"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Description: Detect when an admin role is assigned to a group
DisplayName: "Okta Group Admin Role Assigned"
Enabled: true
Filename: okta_group_admin_role_assigned.py
Reference: https://support.okta.com/help/s/article/How-to-assign-Administrator-roles-to-groups?language=en_US#:~:text=Log%20in%20to%20the%20Admin,user%20and%20click%20Save%20changes
Severity: High
DedupPeriodMinutes: 60
LogTypes:
  - Okta.SystemLog
RuleID: "Okta.Group.Admin.Role.Assigned"
Threshold: 1

AnalysisType: saved_query
QueryName: "Okta HAR File IOCs"
Description: https://sec.okta.com/harfiles
SnowflakeQuery: |-
  SELECT
   *
  FROM
         panther_logs.public.okta_systemlog
  WHERE
         (ARRAYS_OVERLAP(p_any_ip_addresses,ARRAY_CONSTRUCT('23.105.182.19', '104.251.211.122', '202.59.10.100', '162.210.194.35', '198.16.66.124', '198.16.66.156', '198.16.70.28', '198.16.74.203', '198.16.74.204', '198.16.74.205', '198.98.49.203', '2.56.164.52', '207.244.71.82', '207.244.71.84', '207.244.89.161', '207.244.89.162', '23.106.249.52', '23.106.56.11', '23.106.56.21', '23.106.56.36', '23.106.56.37', '23.106.56.38', '23.106.56.54')) OR client:userAgent.rawUserAgent IN ('Mozilla/5.0 (Windows NT 10.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.7113.93 Safari/537.36', ' Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.83 Safari/537.36'))

DatabricksQuery: |-
  SELECT
   *
  FROM
         panther_logs.okta_systemlog
  WHERE
         (SIZE(ARRAY_INTERSECT(p_any_ip_addresses, ARRAY('23.105.182.19', '104.251.211.122', '202.59.10.100', '162.210.194.35', '198.16.66.124', '198.16.66.156', '198.16.70.28', '198.16.74.203', '198.16.74.204', '198.16.74.205', '198.98.49.203', '2.56.164.52', '207.244.71.82', '207.244.71.84', '207.244.89.161', '207.244.89.162', '23.106.249.52', '23.106.56.11', '23.106.56.21', '23.106.56.36', '23.106.56.37', '23.106.56.38', '23.106.56.54'))) > 0 OR client.userAgent.rawUserAgent IN ('Mozilla/5.0 (Windows NT 10.0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.7113.93 Safari/537.36', ' Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.83 Safari/537.36'))

from panther_okta_helpers import okta_alert_context


def rule(event):
    return "system.idp.lifecycle" in event.get("eventType")


def title(event):
    action = event.get("eventType").split(".")[-1]
    target = event.deep_walk(
        "target", "displayName", default="<displayName-not-found>", return_val="first"
    )
    return (
        f"{event.deep_get('actor', 'displayName', default='<displayName-not-found>')} "
        f"<{event.deep_get('actor', 'alternateId', default='alternateId-not-found')}> "
        f"{action}d Identity Provider [{target}]"
    )


def severity(event):
    if "create" in event.get("eventType"):
        return "HIGH"
    return "MEDIUM"


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_idp_create_modify.py
RuleID: "Okta.Identity.Provider.Created.Modified"
DisplayName: "Okta Identity Provider Created or Modified"
Enabled: true
LogTypes:
  - Okta.SystemLog
Reports:
  MITRE ATT&CK:
    - TA0006:T1556 # Modify Authentication Process
    - TA0001:T1199 # Trusted Relationship
    - TA0003:T1098 # Account Manipulation
Severity: High
Description: >
  A new 3rd party Identity Provider has been created or modified.
  Attackers have been observed configuring a second Identity Provider to act as an "impersonation app"
  to access applications within the compromised Org on behalf of other users. This second Identity Provider,
  also controlled by the attacker, would act as a “source” IdP in an inbound federation relationship
  (sometimes called “Org2Org”) with the target.
Runbook: |
  Delegate access to this feature to a Custom Admin Role with the minimum required permissions.
  Constrain these roles to groups that exclude highly privileged administrators.
Reference: >
  https://sec.okta.com/articles/2023/08/cross-tenant-impersonation-prevention-and-detection
DedupPeriodMinutes: 30
Threshold: 1

from panther_okta_helpers import okta_alert_context


def rule(event):
    return event.get("eventType") == "user.authentication.auth_via_IDP"


def title(event):
    target = event.deep_walk(
        "target", "displayName", default="displayName-not-found", return_val="first"
    )
    return (
        f"{event.deep_get('actor', 'displayName', default='<displayName-not-found>')} "
        f"<{event.deep_get('actor', 'alternateId', default='alternateId-not-found')}> "
        f"signed in via 3rd party Identity Provider to {target}"
    )


def alert_context(event):
    return okta_alert_context(event)

AnalysisType: rule
Filename: okta_idp_signin.py
RuleID: "Okta.Identity.Provider.SignIn"
DisplayName: "Okta Identity Provider Sign-in"
Enabled: false
LogTypes:
  - Okta.SystemLog
Tags:
  - Configuration Required
Reports:
  MITRE ATT&CK:
    - TA0001:T1199 # Trusted Relationship
    - TA0003:T1098 # Account Manipulation
Severity: High
Description: >
  A user has signed in using a 3rd party Identity Provider.
  Attackers have been observed configuring a second Identity Provider to act as an "impersonation app"
  to access applications within the compromised Org on behalf of other users. This second Identity Provider,
  also controlled by the attacker, would act as a “source” IdP in an inbound federation relationship
  (sometimes called “Org2Org”) with the target.
  From this “source” IdP, the threat actor manipulated the username parameter for targeted users in the second
  “source” Identity Provider to match a real user in the compromised “target” Identity Provider.
  This provided the ability to Single sign-on (SSO) into applications in the target IdP as the targeted user.
  Do not use this rule if your organization uses legitimate 3rd-party Identity Providers.
Reference: >
  https://sec.okta.com/articles/2023/08/cross-tenant-impersonation-prevention-and-detection
DedupPeriodMinutes: 30
Threshold: 1

AnalysisType: scheduled_query
QueryName: "Okta Investigate MFA and Password resets"
Enabled: false
Description: >
  Investigate Password and MFA resets for the last 7 days
SnowflakeQuery: |
  SELECT p_event_time,actor:alternateId as actor_user,target[0]:alternateId as target_user, eventType,client:ipAddress as ip_address
  FROM panther_logs.public.okta_systemlog
  WHERE eventType IN ('user.mfa.factor.reset_all', 'user.mfa.factor.deactivate', 'user.mfa.factor.suspend', 'user.account.reset_password', 'user.account.update_password','user.mfa.factor.update')
  and p_occurs_since('7 days')
  -- If you wish to investigate an individual user , uncomment this line and add their email here
  -- and actor:alternateId = '<EMAIL_GOES_HERE>'
  ORDER by p_event_time DESC

DatabricksQuery: |
  SELECT p_event_time,actor:alternateId as actor_user,target[0]:alternateId as target_user, eventType,client:ipAddress as ip_address
  FROM panther_logs.okta_systemlog
  WHERE eventType IN ('user.mfa.factor.reset_all', 'user.mfa.factor.deactivate', 'user.mfa.factor.suspend', 'user.account.reset_password', 'user.account.update_password','user.mfa.factor.update')
  and p_occurs_since('7 days')
  -- If you wish to investigate an individual user , uncomment this line and add their email here
  -- and actor:alternateId = '<EMAIL_GOES_HERE>'
  ORDER by p_event_time DESC
Schedule:
  RateMinutes: 43200
  TimeoutMinutes: 1

AnalysisType: scheduled_query
QueryName: "Okta Investigate Session ID Activity"
Enabled: false
Description: >
  Search for activity related to a specific SessionID in Okta panther_logs.okta_systemlog
SnowflakeQuery: |
  SELECT
    p_event_time as event_time,
    actor:alternateId as actor_email,
    actor:displayName as actor_name,
    authenticationContext:externalSessionId as sessionId,
    displayMessage,
    eventType,
    client:ipAddress as src_ip,
    client:geographicalContext:city as city,
    client:geographicalContext:country as country,
    client:userAgent:rawUserAgent as user_agent
  FROM panther_logs.public.okta_systemlog
  WHERE p_occurs_since('7 days')
  -- Uncomment the line below and replace 'sessionId' with the sessionId you are investigating
  -- and authenticationContext:externalSessionId = '<SESSIONID_GOES_HERE>'
  ORDER BY event_time DESC

DatabricksQuery: |
  SELECT
    p_event_time as event_time,
    actor:alternateId as actor_email,
    actor:displayName as actor_name,
    authenticationContext:externalSessionId as sessionId,
    displayMessage,
    eventType,
    client:ipAddress as src_ip,
    client:geographicalContext:city as city,
    client:geographicalContext:country as country,
    client:userAgent:rawUserAgent as user_agent
  FROM panther_logs.okta_systemlog
  WHERE p_occurs_since('7 days')
  -- Uncomment the line below and replace 'sessionId' with the sessionId you are investigating
  -- and authenticationContext:externalSessionId = '<SESSIONID_GOES_HERE>'
  ORDER BY event_time DESC
Schedule:
  RateMinutes: 43200
  TimeoutMinutes: 1

AnalysisType: scheduled_query
QueryName: "Okta Investigate User Activity"
Enabled: false
Description: >
  Audit user activity across your environment. Customize to filter on specific users, time ranges, etc
SnowflakeQuery: |
  SELECT actor:displayName AS actor_name, actor:alternateId AS actor_email, eventType, COUNT(*) AS activity_count
  FROM panther_logs.public.okta_systemlog
  WHERE p_occurs_since('7 days')
  AND actor:type = 'User'
  -- Uncomment lines below to filter by user email and/or eventType
  -- and actor_email = 'email'
  -- and eventType = 'eventType'
  GROUP BY actor:displayName, actor:alternateId, eventType
  ORDER BY  actor_name, activity_count DESC

DatabricksQuery: |
  SELECT actor:displayName AS actor_name, actor:alternateId AS actor_email, eventType, COUNT(*) AS activity_count
  FROM panther_logs.okta_systemlog
  WHERE p_occurs_since('7 days')
  AND actor:type = 'User'
  -- Uncomment lines below to filter by user email and/or eventType
  -- and actor_email = 'email'
  -- and eventType = 'eventType'
  GROUP BY actor:displayName, actor:alternateId, eventType
  ORDER BY  actor_name, activity_count DESC
Schedule:
  RateMinutes: 43200
  TimeoutMinutes: 1

def rule(_):
    return True


def title(event):
    return (
        "Okta Login for "
        f"[{event.deep_get('actor', 'alternateId', default = '<email_not_found>')}]"
        " from unmanaged IP Address."
    )

AnalysisType: scheduled_rule
Description: Detects Okta Logins from IP addresses not found in CrowdStrike''s AIP list. May indicate unmanaged device being used, or faulty CrowdStrike Sensor.
DisplayName: "Okta Login From CrowdStrike Unmanaged Device"
Enabled: false
Filename: okta_login_from_crowdstrike_unmanaged_device.py
Reference: https://www.crowdstrike.com/wp-content/uploads/2023/05/crowdstrike-falcon-device-control-data-sheet.pdf
Severity: Medium
DedupPeriodMinutes: 60
RuleID: "Okta.Login.From.CrowdStrike.Unmanaged.Device"
Threshold: 1
ScheduledQueries:
  - Okta Login From CrowdStrike Unmanaged Device
Tags:
  - Multi-Table Query

AnalysisType: scheduled_query
Description: Okta Logins from an IP Address not found in CrowdStrike's AIP List
Enabled: false
SnowflakeQuery: |
  SELECT *
  FROM panther_logs.public.okta_systemlog
  WHERE p_occurs_since('1 hour')
    AND eventtype = 'user.session.start'
    AND outcome:result = 'SUCCESS'
    AND client:device = 'Computer'
    AND client:ipAddress LIKE '%.%.%.%'
    AND client:ipAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.public.crowdstrike_aidmaster
        WHERE p_occurs_since('3 days')
      )

DatabricksQuery: |
  SELECT *
  FROM panther_logs.okta_systemlog
  WHERE p_occurs_since('1 hour')
    AND eventtype = 'user.session.start'
    AND outcome:result = 'SUCCESS'
    AND client:device = 'Computer'
    AND client:ipAddress LIKE '%.%.%.%'
    AND client:ipAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.crowdstrike_aidmaster
        WHERE p_occurs_since('3 days')
      )
QueryName: "Okta Login From CrowdStrike Unmanaged Device"
Schedule:
  RateMinutes: 60
  TimeoutMinutes: 1
Tags:
  - Multi-Table Query

# This file is the part of the Crowdstrike FDREvent migration, and it's the equivalent of
# https://github.com/panther-labs/panther-analysis/blob/b61db1ecf3967c5f6a44c1782f8891fd5f54384d/queries/okta_queries/Okta_Login_From_CrowdStrike_Unmanaged_Device.yml
#
AnalysisType: scheduled_query
Description: Okta Logins from an IP Address not found in CrowdStrike's AIP List (crowdstrike_fdrevent table)
Enabled: false
SnowflakeQuery: |
  SELECT *
  FROM panther_logs.public.okta_systemlog
  WHERE p_occurs_since('1 days')
    AND eventtype = 'user.session.start'
    AND outcome:result = 'SUCCESS'
    AND client:device = 'Computer'
    AND client:ipAddress LIKE '%.%.%.%'
    AND client:ipAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.public.crowdstrike_fdrevent
        WHERE p_occurs_since('3 days') AND panther_logs.public.crowdstrike_fdrevent.fdr_event_type = 'aid_master'
      )

DatabricksQuery: |
  SELECT *
  FROM panther_logs.okta_systemlog
  WHERE p_occurs_since('1 days')
    AND eventtype = 'user.session.start'
    AND outcome:result = 'SUCCESS'
    AND client:device = 'Computer'
    AND client:ipAddress LIKE '%.%.%.%'
    AND client:ipAddress NOT IN
      (
        SELECT DISTINCT aip
        FROM panther_logs.crowdstrike_fdrevent
        WHERE p_occurs_since('3 days') AND panther_logs.crowdstrike_fdrevent.fdr_event_type = 'aid_master'
      )
QueryName: "Okta Login From CrowdStrike Unmanaged Device (crowdstrike_fdrevent table)"
Schedule:
  RateMinutes: 1440
  TimeoutMinutes: 1
Tags:
  - Multi-Table Query

def rule(event):
    return (
        event.get("eventType") == "user.session.start"
        and event.deep_get("outcome", "result") == "SUCCESS"
    )


def title(event):
    return f'{event.deep_get("actor", "displayName")} logged in to Okta'

AnalysisType: rule
Filename: okta_login_signal.py
RuleID: "Okta.Login.Success"
DisplayName: "Okta Login Signal"
Enabled: false
CreateAlert: false
LogTypes:
  - Okta.SystemLog
Severity: Info
DedupPeriodMinutes: 60
Threshold: 1

AnalysisType: correlation_rule
RuleID: "Okta.Login.Without.Push.Group"
DisplayName: "Okta Login Without Push"
Enabled: false
Tags:
  - Okta
  - Push Security
  - Identity Verification
  - Configuration Required
  - Credential Access
Reports:
  MITRE ATT&CK:
    - TA0001:T1078 # Initial Access: Valid Accounts
    - TA0006:T1539 # Credential Access: Steal Web Session Cookie
    - TA0006:T1621 # Credential Access: Multi-Factor Authentication Request Generation
Severity: Critical
Description: >
  Identifies successful Okta logins not followed by Push Security authorization within 60 minutes. Push Security provides additional identity verification beyond Okta MFA as a defense-in-depth strategy. Missing Push Security verification suggests compromised credentials, session hijacking, or MFA bypass where attackers satisfied Okta authentication but cannot complete additional verification.
Runbook: |
  1. Query Okta System Log for all authentication events by actor.alternateId in the 90 minutes around the login to check if Push Security authentication occurred outside the 60-minute detection window, and review the source IP, geolocation, device, and MFA method used
  2. Query Push Security logs for any authentication attempts or failures by the same user in the 2 hours around the Okta login to determine if the user attempted but failed to complete Push Security verification
  3. Check Okta audit logs for all application access, permission changes, and administrative actions during the Okta session to identify suspicious activity that may indicate compromised credentials or session hijacking
Reference: https://www.okta.com/resources/datasheet/okta-adaptive-multi-factor-authentication-product-datasheet/
Detection:
  - Group:
      - ID: Push
        RuleID: Push.Security.Authorized.IdP.Login
        Absence: true
      - ID: Okta
        RuleID: Okta.Login.Success
    MatchCriteria:
      field_name:
        - GroupID: Push
          Match: new.email
        - GroupID: Okta
          Match: actor.alternateId
    Schedule:
      RateMinutes: 1440
      TimeoutMinutes: 10
    LookbackWindowMinutes: 1800

# configure this Push marker based on your environment
PUSH_MARKER = "PS_mxzqarw"


def rule(event):
    return not event.deep_get("client", "userAgent", "rawUserAgent", default="").endswith(
        PUSH_MARKER
    )


def title(event):
    actor = event.deep_get("actor", "displayName")
    return f"{actor} logged in from device without expected Push marker"

AnalysisType: rule
Filename: okta_login_without_push_marker.py
RuleID: "Okta.Login.Without.Push.Marker"
DisplayName: "Okta Login Without Push Marker"
Enabled: false
Tags:
  - Push Security
  - Configuration Required
LogTypes:
  - Okta.SystemLog
Severity: Medium
DedupPeriodMinutes: 60
Threshold: 1

import panther_event_type_helpers as event_type


def rule(event):
    return event.udm("event_type") == event_type.ADMIN_MFA_DISABLED


def title(event):
    return f"Okta System-wide MFA Disabled by Admin User {event.udm('actor_user')}"


def alert_context(event):
    context = {
        "user": event.udm("actor_user"),
        "ip": event.udm("source_ip"),
        "event": event.get("eventType"),
    }
    return context

AnalysisType: rule
Filename: okta_admin_disabled_mfa.py
RuleID: "Okta.Global.MFA.Disabled"
DisplayName: "Okta MFA Globally Disabled"
Enabled: true
LogTypes:
  - Okta.SystemLog
Tags:
  - Identity & Access Management
  - DataModel
  - Okta
  - Defense Evasion:Modify Authentication Process
Reports:
  MITRE ATT&CK:
    - TA0005:T1556
Severity: High
Description: An admin user has disabled the MFA requirement for your Okta account
Reference: https://help.okta.com/oie/en-us/content/topics/identity-engine/authenticators/about-authenticators.htm
Runbook: Contact Admin to ensure this was sanctioned activity
DedupPeriodMinutes: 15
SummaryAttributes:
  - eventType
  - severity
  - displayMessage
  - p_any_ip_addresses

import json

from panther_base_helpers import deep_get
from panther_okta_helpers import okta_alert_context


def rule(event):
    if event.get("eventtype") != "policy.evaluate_sign_on":
        return False

    if "Okta Admin Console" not in event.deep_walk("target", "displayName", default=""):
        return False

    behaviors = event.deep_get("debugContext", "debugData", "behaviors")
    if behaviors:
        return "New Device=POSITIVE" in behaviors and "New IP=POSITIVE" in behaviors

    log_only_security_data = event.deep_get("debugContext", "debugData", "logOnlySecurityData")
    if isinstance(log_only_security_data, str):
        log_only_security_data = json.loads(log_only_security_data)
    return (
        deep_get(log_only_security_data, "behaviors", "New Device") == "POSITIVE"
        and deep_get(log_only_security_data, "behaviors", "New IP") == "POSITIVE"
    )


def title(event):
    return (
        f"{event.deep_get('actor', 'displayName', default='<displayName-not-found>')} "
        f"<{event.deep_get('actor', 'alternateId', default='alternateId-not-found')}> "
        f"accessed Okta Admin Console using new behaviors: "
        f"New IP: {event.deep_get('client', 'ipAddress', default='<ipAddress-not-found>')} "
        f"New Device: {event.deep_get('device', 'name', default='<deviceName-not-found>')}"
    )


def alert_context(event):
    return okta_alert_context(event)