Panther rules: otx

from panther_otx_helpers import (
    get_otx_object,
    otx_alert_context,
    otx_severity,
    severity_greater_than,
)

INDICATOR_FIELDS = (
    "p_any_ip_addresses",
    "p_any_domain_names",
    "p_any_md5_hashes",
    "p_any_sha1_hashes",
    "p_any_sha256_hashes",
    "p_any_emails",
)

MATCHED_INDICATORS = {}  # {indicator: indicator_type}


def rule(event):
    global MATCHED_INDICATORS  # pylint: disable=global-statement
    MATCHED_INDICATORS = {}

    otx = get_otx_object(event)
    if not otx:
        return False

    for field in INDICATOR_FIELDS:
        for value in event.get(field, []) or []:
            if value in MATCHED_INDICATORS:
                continue
            indicator_type = otx.indicator_type(value)
            if not indicator_type:
                continue
            MATCHED_INDICATORS[value] = indicator_type

    return bool(MATCHED_INDICATORS)


def title(event):
    log_type = event.get("p_log_type", "Unknown")
    if len(MATCHED_INDICATORS) == 1:
        indicator, ioc_type = next(iter(MATCHED_INDICATORS.items()))
        return f"OTX: Known threat {ioc_type} [{indicator}] detected in {log_type}"
    return f"OTX: {len(MATCHED_INDICATORS)} threat indicators detected in {log_type}"


def severity(event):
    highest = None
    for indicator in MATCHED_INDICATORS:
        sev = otx_severity(event, indicator)
        if highest is None or severity_greater_than(sev, highest):
            highest = sev
    return highest or "DEFAULT"


def alert_context(event):
    if not MATCHED_INDICATORS:
        return {}
    ctx = {}
    for indicator, indicator_type in MATCHED_INDICATORS.items():
        indicator_ctx = otx_alert_context(event, indicator)
        indicator_ctx["MatchedIndicatorType"] = indicator_type
        ctx[indicator] = indicator_ctx
    return ctx

AnalysisType: rule
Filename: otx_malicious_indicator.py
RuleID: "Standard.OTX.MaliciousIndicator"
DisplayName: "OTX Threat Intelligence Indicator Match"
Enabled: true
Severity: High
Description: >-
  Detects when an IP address in any log event matches a known threat indicator
  from AlienVault OTX pulse intelligence. Severity is elevated when the pulse
  includes a named adversary or known malware families.
Runbook: |
  1. Review the alert context and open the OTX Pulse URL to assess the indicator's
     threat context, associated adversary, malware families, and MITRE ATT&CK mappings.
  2. Query the data lake for all events involving this indicator to determine what assets
     or services were contacted and whether any connections were successful.
  3. If the indicator is confirmed malicious and interaction was observed, block the
     indicator, isolate affected hosts, and reset any credentials that may have been exposed.
Reference: https://otx.alienvault.com
DedupPeriodMinutes: 60
Reports:
  MITRE ATT&CK:
    - TA0043:T1595.001
Tags:
  - Reconnaissance:Active Scanning
  - OTX
  - Threat Intelligence
SummaryAttributes:
  - p_any_ip_addresses
  - p_source_label
LogTypes:
  - Amazon.EKS.Audit
  - Asana.Audit
  - Atlassian.Audit
  - AWS.ALB
  - AWS.CloudTrail
  - AWS.VPCFlow
  - Azure.Audit
  - Azure.MonitorActivity
  - Box.Event
  - Cloudflare.Firewall
  - Cloudflare.HttpRequest
  - Crowdstrike.FDREvent
  - GCP.AuditLog
  - GSuite.ActivityEvent
  - Notion.AuditLogs
  - Okta.SystemLog
  - OneLogin.Events
  - OnePassword.SignInAttempt
  - Zendesk.Audit
  - Zoom.Activity

AnalysisType: saved_query
QueryName: "Query.OTX.HighImpactPulses"
Description: >
  Identify high-impact OTX pulses by grouping per-indicator rows and surfacing
  pulses with a named adversary or known malware families. Each pulse is scored
  CRITICAL (adversary AND malware), HIGH (one of the two), or filtered out.
  Results are ordered by severity and indicator volume to highlight the most
  noteworthy threats reported recently.
Tags:
  - OTX
  - Threat Intelligence
  - Lookup Table
SnowflakeQuery: |
  WITH pulse_summary AS (
      SELECT
          id AS pulse_id,
          ANY_VALUE(name) AS pulse_name,
          ANY_VALUE(description) AS description,
          ANY_VALUE(adversary) AS adversary,
          ANY_VALUE(malware_families) AS malware_families,
          ANY_VALUE(industries) AS industries,
          ANY_VALUE(target_countries) AS target_countries,
          ANY_VALUE(attack_ids) AS attack_ids,
          ANY_VALUE(tags) AS tags,
          ANY_VALUE(tlp) AS tlp,
          ANY_VALUE(modified) AS pulse_modified,
          COUNT(*) AS indicator_count,
          ARRAY_SIZE(ARRAY_AGG(DISTINCT indicator_type)) AS distinct_indicator_types,
          ARRAY_AGG(DISTINCT indicator_type) WITHIN GROUP (ORDER BY indicator_type) AS indicator_types
      FROM panther_lookups.public.otx_pulses
      WHERE p_occurs_since('30 days')
      GROUP BY id
  )
  SELECT
      pulse_id,
      pulse_name,
      description,
      adversary,
      malware_families,
      industries,
      target_countries,
      attack_ids,
      tags,
      tlp,
      indicator_count,
      distinct_indicator_types,
      indicator_types,
      pulse_modified,
      CASE
          WHEN adversary IS NOT NULL AND adversary <> '' AND ARRAY_SIZE(malware_families) > 0
              THEN 'CRITICAL'
          ELSE 'HIGH'
      END AS pulse_severity,
      CONCAT('https://otx.alienvault.com/pulse/', pulse_id) AS otx_url
  FROM pulse_summary
  WHERE
      (adversary IS NOT NULL AND adversary <> '')
      OR ARRAY_SIZE(malware_families) > 0
  ORDER BY pulse_severity, indicator_count DESC, pulse_modified DESC
  LIMIT 100

DatabricksQuery: |
  WITH pulse_summary AS (
      SELECT
          id AS pulse_id,
          ANY_VALUE(name) AS pulse_name,
          ANY_VALUE(description) AS description,
          ANY_VALUE(adversary) AS adversary,
          ANY_VALUE(malware_families) AS malware_families,
          ANY_VALUE(industries) AS industries,
          ANY_VALUE(target_countries) AS target_countries,
          ANY_VALUE(attack_ids) AS attack_ids,
          ANY_VALUE(tags) AS tags,
          ANY_VALUE(tlp) AS tlp,
          ANY_VALUE(modified) AS pulse_modified,
          COUNT(*) AS indicator_count,
          SIZE(COLLECT_SET(indicator_type)) AS distinct_indicator_types,
          ARRAY_SORT(COLLECT_SET(indicator_type)) AS indicator_types
      FROM panther_lookups.otx_pulses
      WHERE p_occurs_since('30 days')
      GROUP BY id
  )
  SELECT
      pulse_id,
      pulse_name,
      description,
      adversary,
      malware_families,
      industries,
      target_countries,
      attack_ids,
      tags,
      tlp,
      indicator_count,
      distinct_indicator_types,
      indicator_types,
      pulse_modified,
      CASE
          WHEN adversary IS NOT NULL AND adversary <> '' AND SIZE(malware_families) > 0
              THEN 'CRITICAL'
          ELSE 'HIGH'
      END AS pulse_severity,
      CONCAT('https://otx.alienvault.com/pulse/', pulse_id) AS otx_url
  FROM pulse_summary
  WHERE
      (adversary IS NOT NULL AND adversary <> '')
      OR SIZE(malware_families) > 0
  ORDER BY pulse_severity, indicator_count DESC, pulse_modified DESC
  LIMIT 100

AnalysisType: saved_query
QueryName: "Query.OTX.PulsesSummary"
Description: >
  Group OTX pulse rows by pulse and summarize the indicators each pulse contains.
  The OTX.Pulses lookup table stores one indicator per row; this query reconstructs
  the pulse view with counts by indicator type, adversary, malware families,
  industries, and target countries.
Tags:
  - OTX
  - Threat Intelligence
  - Lookup Table
SnowflakeQuery: |
  SELECT
      id AS pulse_id,
      ANY_VALUE(name) AS pulse_name,
      ANY_VALUE(adversary) AS adversary,
      ANY_VALUE(malware_families) AS malware_families,
      ANY_VALUE(tags) AS tags,
      ANY_VALUE(industries) AS industries,
      ANY_VALUE(target_countries) AS target_countries,
      ANY_VALUE(attack_ids) AS attack_ids,
      ANY_VALUE(tlp) AS tlp,
      ANY_VALUE(created) AS pulse_created,
      ANY_VALUE(modified) AS pulse_modified,
      COUNT(*) AS indicator_count,
      ARRAY_AGG(DISTINCT indicator_type) WITHIN GROUP (ORDER BY indicator_type) AS indicator_types,
      COUNT_IF(indicator_type ILIKE 'IPv4' OR indicator_type ILIKE 'IPv6') AS ip_count,
      COUNT_IF(indicator_type ILIKE 'domain' OR indicator_type ILIKE 'hostname') AS domain_count,
      COUNT_IF(indicator_type ILIKE 'FileHash-%') AS hash_count,
      COUNT_IF(indicator_type ILIKE 'URL' OR indicator_type ILIKE 'URI') AS url_count,
      COUNT_IF(indicator_type ILIKE 'email') AS email_count,
      COUNT_IF(indicator_type ILIKE 'CVE') AS cve_count,
      CONCAT('https://otx.alienvault.com/pulse/', id) AS otx_url
  FROM panther_lookups.public.otx_pulses
  WHERE p_occurs_since('30 days')
  GROUP BY id
  ORDER BY pulse_modified DESC
  LIMIT 200

DatabricksQuery: |
  SELECT
      id AS pulse_id,
      ANY_VALUE(name) AS pulse_name,
      ANY_VALUE(adversary) AS adversary,
      ANY_VALUE(malware_families) AS malware_families,
      ANY_VALUE(tags) AS tags,
      ANY_VALUE(industries) AS industries,
      ANY_VALUE(target_countries) AS target_countries,
      ANY_VALUE(attack_ids) AS attack_ids,
      ANY_VALUE(tlp) AS tlp,
      ANY_VALUE(created) AS pulse_created,
      ANY_VALUE(modified) AS pulse_modified,
      COUNT(*) AS indicator_count,
      ARRAY_SORT(COLLECT_SET(indicator_type)) AS indicator_types,
      COUNT_IF(indicator_type ILIKE 'IPv4' OR indicator_type ILIKE 'IPv6') AS ip_count,
      COUNT_IF(indicator_type ILIKE 'domain' OR indicator_type ILIKE 'hostname') AS domain_count,
      COUNT_IF(indicator_type ILIKE 'FileHash-%') AS hash_count,
      COUNT_IF(indicator_type ILIKE 'URL' OR indicator_type ILIKE 'URI') AS url_count,
      COUNT_IF(indicator_type ILIKE 'email') AS email_count,
      COUNT_IF(indicator_type ILIKE 'CVE') AS cve_count,
      CONCAT('https://otx.alienvault.com/pulse/', id) AS otx_url
  FROM panther_lookups.otx_pulses
  WHERE p_occurs_since('30 days')
  GROUP BY id
  ORDER BY pulse_modified DESC
  LIMIT 200