Implementing Soar Automation With Phantom

Implementing SOAR Automation with Phantom

When to Use

Use this skill when:

• SOC teams need to automate repetitive triage and enrichment tasks for high-volume alerts
• Manual response times exceed SLA requirements and automation can reduce MTTR
• Multiple security tools (SIEM, EDR, firewall, TIP) need orchestrated response actions
• Playbook standardization is required to ensure consistent analyst response across shifts

Do not use for fully autonomous containment without human approval gates — always include analyst decision points for high-impact actions like account disabling or host isolation.

Common Misconfigurations & Verification

• Playbook never triggers: the most common failure is a trigger/label mismatch — the playbook is scoped to label: ["phishing","notable"] and severity: ["high","medium"], but Splunk ES sends containers with a different label (e.g., events or a custom label) or severity casing, so it ingests but no playbook runs. Verify on the container itself that the label/severity match the trigger, and that the playbook is set active: true and run_as an account with action permissions.
• phantom.act silently no-ops on bad asset/action name: action names are app-specific ("url reputation" vs "url_reputation", "quarantine device" vs "contain device") and must match the configured asset exactly; a typo or wrong assets=[...] name fails the action and the callback chain stalls. Check the playbook run's debug/audit log shows each action success, not just that the playbook "completed."
• Artifact CEF path mismatch: the playbooks read cef.sourceAddress, cef.type, cef.destinationDnsDomain — if ES maps the IOC to a different CEF field, the for artifact loop finds nothing and enrichment/containment never queue. Inspect a real container's artifact CEF keys and align.
• Approval gate that auto-expires into inaction: phantom.prompt(respond_in_mins=15) with no timeout branch means a missed prompt leaves the host un-contained with no fallback — define the no-response path explicitly.
• Verification: fire a synthetic test container with known artifacts through the playbook in a non-prod tenant, confirm each action executes against a sandbox asset (VT lookup returns, firewall block logs, prompt reaches the analyst queue), and confirm a clean IOC auto-closes while a malicious one escalates — before enabling on live notables.

Prerequisites

• Splunk SOAR (Phantom) 6.x+ deployed with web interface access
• App connectors configured: VirusTotal, CrowdStrike, ServiceNow, Active Directory, Splunk ES
• Splunk ES integration for ingesting notable events as SOAR events
• API credentials for each integrated tool stored in SOAR asset configuration
• Python knowledge for custom playbook actions

Workflow

Step 1: Configure Asset Connections

Set up integrations with security tools via SOAR Apps:

VirusTotal Asset Configuration:

{
  "app": "VirusTotal v3",
  "asset_name": "virustotal_prod",
  "configuration": {
    "api_key": "YOUR_VT_API_KEY",
    "rate_limit": true,
    "max_requests_per_minute": 4
  },
  "product_vendor": "VirusTotal",
  "product_name": "VirusTotal"
}

CrowdStrike Falcon Asset:

{
  "app": "CrowdStrike Falcon",
  "asset_name": "crowdstrike_prod",
  "configuration": {
    "client_id": "CS_CLIENT_ID",
    "client_secret": "CS_CLIENT_SECRET",
    "base_url": "https://api.crowdstrike.com"
  }
}

Active Directory Asset:

{
  "app": "Active Directory",
  "asset_name": "ad_prod",
  "configuration": {
    "server": "dc01.company.com",
    "username": "soar_service@company.com",
    "password": "SERVICE_ACCOUNT_PASSWORD",
    "ssl": true
  }
}

Step 2: Build Phishing Triage Playbook

Create an automated phishing response playbook in Python (Phantom playbook format):

"""
Phishing Triage Automation Playbook
Trigger: New phishing email reported via Splunk ES notable or email ingestion
"""

import phantom.rules as phantom
import json

def on_start(container):
    # Extract artifacts (URLs, file hashes, sender) from the container
    artifacts = phantom.get_artifacts(container_id=container["id"])

    for artifact in artifacts:
        artifact_type = artifact.get("cef", {}).get("type", "")

        if artifact_type == "url":
            phantom.act("url reputation", targets=artifact,
                        assets=["virustotal_prod"],
                        callback=url_reputation_callback,
                        name="url_reputation")

        elif artifact_type == "hash":
            phantom.act("file reputation", targets=artifact,
                        assets=["virustotal_prod"],
                        callback=hash_reputation_callback,
                        name="file_reputation")

        elif artifact_type == "ip":
            phantom.act("ip reputation", targets=artifact,
                        assets=["virustotal_prod"],
                        callback=ip_reputation_callback,
                        name="ip_reputation")

def url_reputation_callback(action, success, container, results, handle):
    if not success:
        phantom.comment(container, "URL reputation check failed")
        return

    for result in results:
        data = result.get("data", [{}])[0]
        malicious_count = data.get("summary", {}).get("malicious", 0)
        total_engines = data.get("summary", {}).get("total_engines", 0)

        if malicious_count > 5:
            # High confidence malicious — auto-block and escalate
            phantom.act("block url", targets=result,
                        assets=["palo_alto_prod"],
                        name="block_malicious_url")

            phantom.set_severity(container, "high")
            phantom.set_status(container, "open")
            phantom.comment(container,
                f"URL flagged by {malicious_count}/{total_engines} engines. "
                f"Blocked on firewall. Escalating to Tier 2.")

            # Create ServiceNow ticket
            phantom.act("create ticket", targets=container,
                        assets=["servicenow_prod"],
                        parameters=[{
                            "short_description": f"Phishing - Malicious URL detected",
                            "urgency": "2",
                            "impact": "2"
                        }],
                        name="create_incident_ticket")

        elif malicious_count > 0:
            # Medium confidence — request analyst review
            phantom.promote(container, template="Phishing Investigation")
            phantom.comment(container,
                f"URL flagged by {malicious_count}/{total_engines} engines. "
                f"Requires analyst review.")

        else:
            # Clean — close with comment
            phantom.set_status(container, "closed")
            phantom.comment(container,
                f"URL clean: 0/{total_engines} engines flagged. Auto-closed.")

def hash_reputation_callback(action, success, container, results, handle):
    if not success:
        return

    for result in results:
        data = result.get("data", [{}])[0]
        positives = data.get("summary", {}).get("positives", 0)

        if positives > 10:
            # Known malware — quarantine and block
            phantom.act("quarantine device", targets=result,
                        assets=["crowdstrike_prod"],
                        name="isolate_endpoint")
            phantom.set_severity(container, "high")

def ip_reputation_callback(action, success, container, results, handle):
    if not success:
        return

    for result in results:
        data = result.get("data", [{}])[0]
        malicious = data.get("summary", {}).get("malicious", 0)

        if malicious > 3:
            phantom.act("block ip", targets=result,
                        assets=["palo_alto_prod"],
                        name="block_malicious_ip")

Step 3: Build Alert Enrichment Playbook

Automate enrichment for all incoming SIEM alerts:

"""
Universal Alert Enrichment Playbook
Runs on every new event to add context before analyst review
"""

import phantom.rules as phantom

def on_start(container):
    # Get all artifacts
    success, message, artifacts = phantom.get_artifacts(
        container_id=container["id"], full_data=True
    )

    ip_artifacts = [a for a in artifacts if a.get("cef", {}).get("sourceAddress")]
    domain_artifacts = [a for a in artifacts if a.get("cef", {}).get("destinationDnsDomain")]

    # Enrich IPs in parallel
    for artifact in ip_artifacts:
        ip = artifact["cef"]["sourceAddress"]

        # VirusTotal lookup
        phantom.act("ip reputation",
                    parameters=[{"ip": ip}],
                    assets=["virustotal_prod"],
                    callback=enrich_ip_callback,
                    name=f"vt_ip_{ip}")

        # GeoIP lookup
        phantom.act("geolocate ip",
                    parameters=[{"ip": ip}],
                    assets=["maxmind_prod"],
                    callback=geoip_callback,
                    name=f"geo_{ip}")

        # Whois lookup
        phantom.act("whois ip",
                    parameters=[{"ip": ip}],
                    assets=["whois_prod"],
                    name=f"whois_{ip}")

    # Enrich domains
    for artifact in domain_artifacts:
        domain = artifact["cef"]["destinationDnsDomain"]
        phantom.act("domain reputation",
                    parameters=[{"domain": domain}],
                    assets=["virustotal_prod"],
                    name=f"vt_domain_{domain}")

def enrich_ip_callback(action, success, container, results, handle):
    """Update container with enrichment data"""
    if success:
        for result in results:
            summary = result.get("summary", {})
            phantom.add_artifact(container, {
                "cef": {
                    "vt_malicious": summary.get("malicious", 0),
                    "vt_suspicious": summary.get("suspicious", 0),
                    "enrichment_source": "VirusTotal"
                },
                "label": "enrichment",
                "name": "VT IP Enrichment"
            })

Step 4: Implement Approval Gates for High-Impact Actions

Add human-in-the-loop for critical actions:

def containment_decision(action, success, container, results, handle):
    """Present analyst with containment options"""
    phantom.prompt(
        container=container,
        user="soc_tier2",
        message=(
            "Confirmed malicious activity detected.\n"
            f"Host: {container['artifacts'][0]['cef'].get('sourceAddress')}\n"
            f"Threat: {results[0]['summary'].get('threat_name')}\n\n"
            "Select containment action:"
        ),
        respond_in_mins=15,
        options=["Isolate Host", "Disable Account", "Both", "Monitor Only"],
        callback=execute_containment
    )

def execute_containment(action, success, container, results, handle):
    response = results.get("response", "Monitor Only")

    if response in ["Isolate Host", "Both"]:
        phantom.act("quarantine device",
                    parameters=[{"hostname": container["artifacts"][0]["cef"]["sourceHostName"]}],
                    assets=["crowdstrike_prod"],
                    name="isolate_host")

    if response in ["Disable Account", "Both"]:
        phantom.act("disable user",
                    parameters=[{"username": container["artifacts"][0]["cef"]["sourceUserName"]}],
                    assets=["ad_prod"],
                    name="disable_account")

    phantom.comment(container, f"Analyst approved: {response}")

Step 5: Configure Playbook Scheduling and Triggers

Set up event triggers in SOAR:

{
  "playbook_name": "phishing_triage_automation",
  "trigger": {
    "type": "event_created",
    "conditions": {
      "label": ["phishing", "notable"],
      "severity": ["high", "medium"]
    }
  },
  "active": true,
  "run_as": "automation_user"
}

Step 6: Monitor Playbook Performance

Track automation effectiveness with SOAR metrics:

# Query SOAR API for playbook execution stats
import requests

headers = {"ph-auth-token": "YOUR_SOAR_TOKEN"}
response = requests.get(
    "https://soar.company.com/rest/playbook_run",
    headers=headers,
    params={
        "page_size": 100,
        "filter": '{"status":"success"}',
        "sort": "create_time",
        "order": "desc"
    }
)
runs = response.json()["data"]

# Calculate automation metrics
total_runs = len(runs)
avg_duration = sum(r["end_time"] - r["start_time"] for r in runs) / total_runs
auto_closed = sum(1 for r in runs if r.get("auto_resolved"))
print(f"Total runs: {total_runs}")
print(f"Avg duration: {avg_duration:.1f}s")
print(f"Auto-resolved: {auto_closed}/{total_runs} ({auto_closed/total_runs*100:.0f}%)")

Key Concepts

Tools & Systems

• Splunk SOAR (Phantom): Enterprise SOAR platform with 300+ app integrations and visual playbook editor
• Splunk ES: SIEM platform feeding notable events into SOAR as containers for automated triage
• CrowdStrike Falcon: EDR platform integrated via SOAR for automated host isolation and threat hunting
• ServiceNow: ITSM platform integrated for automated incident ticket creation and tracking
• Palo Alto NGFW: Firewall integrated for automated IP/URL blocking via SOAR playbooks

Common Scenarios

• Phishing Triage: Auto-extract URLs/attachments, detonate in sandbox, block malicious, create ticket
• Malware Alert Enrichment: Auto-enrich file hashes across VT/MalwareBazaar, isolate if confirmed malicious
• Brute Force Response: Auto-check if attack succeeded, disable account if compromised, block source IP
• Threat Intel IOC Processing: Auto-ingest TI feed IOCs, check against internal logs, create blocks for matches
• Vulnerability Alert Response: Auto-query asset database for affected systems, create patching ticket with priority

Output Format

SOAR PLAYBOOK EXECUTION REPORT
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Playbook:     Phishing Triage Automation v2.3
Container:    SOAR-2024-08921
Trigger:      Notable event from Splunk ES (phishing)

Actions Executed:
  [1] URL Reputation (VirusTotal)     — 14/90 engines malicious    [2.1s]
  [2] IP Reputation (AbuseIPDB)       — Confidence: 85%            [1.3s]
  [3] Block URL (Palo Alto)           — Blocked on PA-5260         [0.8s]
  [4] Block IP (Palo Alto)            — Blocked on PA-5260         [0.7s]
  [5] Create Ticket (ServiceNow)      — INC0012345 created         [1.5s]
  [6] Prompt Analyst (Tier 2)         — Response: "Isolate Host"   [4m 12s]
  [7] Quarantine Device (CrowdStrike) — WORKSTATION-042 isolated   [3.2s]

Total Duration:    4m 22s (vs 35min avg manual triage)
Time Saved:        ~31 minutes
Disposition:       True Positive — Escalated to IR