Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

MITRE ATT&CK

T1078 T1566 T1685.002 T1685.005

NIST CSF 2.0

DE.AE-02 DE.AE-06 DE.CM-01 RS.MA-01

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.

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

Term	Definition
SOAR	Security Orchestration, Automation, and Response — platform integrating security tools with automated playbooks
Playbook	Automated workflow defining sequential and parallel actions triggered by security events
Asset	SOAR configuration for a connected security tool (API endpoint, credentials, connection parameters)
Container	SOAR event object containing artifacts (IOCs) from an ingested alert or incident
Artifact	Individual IOC or data point within a container (IP, hash, URL, domain, email)
Approval Gate	Human-in-the-loop step requiring analyst decision before executing high-impact automated actions

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

Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.6 KB

API Reference: Implementing SOAR Automation with Phantom

Libraries

requests (HTTP Client for SOAR REST API)

Install: pip install requests
Authentication: ph-auth-token header with API token

Splunk SOAR REST API

Playbooks

Endpoint	Method	Description
`/rest/playbook`	GET	List all playbooks
`/rest/playbook/{id}`	GET	Get playbook details
`/rest/playbook_run`	POST	Execute a playbook

Containers (Events/Incidents)

Endpoint	Method	Description
`/rest/container`	GET	List containers
`/rest/container`	POST	Create new container
`/rest/container/{id}`	GET	Get container details
`/rest/container/{id}`	POST	Update container

Artifacts (IOCs)

Endpoint	Method	Description
`/rest/artifact`	POST	Add artifact to container
`/rest/artifact/{id}`	GET	Get artifact details
CEF fields: `sourceAddress`, `destinationAddress`, `fileHash`, `fileName`

Actions

Endpoint	Method	Description
`/rest/action_run`	POST	Run an action on an asset
`/rest/action_run/{id}`	GET	Get action results
`/rest/app`	GET	List installed apps
`/rest/asset`	GET	List configured assets

System

Endpoint	Method	Description
`/rest/system_info`	GET	System version and status
`/rest/ph_user`	GET	List SOAR users

Common App Actions

App	Action	Description
VirusTotal	`file_reputation`	Check hash reputation
VirusTotal	`url_reputation`	Check URL safety
CrowdStrike	`contain_device`	Network isolate host
ActiveDirectory	`disable_user`	Disable AD account
ServiceNow	`create_ticket`	Create incident ticket
Exchange	`quarantine_email`	Remove phishing email
Splunk	`run_query`	Execute SPL search

Playbook Types

Automation: Fully automated, no analyst input
Investigation: Enrichment with analyst decision gates
Response: Containment actions with approval prompts
Reporting: Data collection and notification

External References

SOAR REST API: https://docs.splunk.com/Documentation/SOAR/current/PlatformAPI/
Playbook Guide: https://docs.splunk.com/Documentation/SOAR/current/DevelopPlaybooks/
App Development: https://docs.splunk.com/Documentation/SOAR/current/DevelopApps/
Splunkbase Apps: https://splunkbase.splunk.com/apps/#/product/soar

Scripts 1

agent.py9.4 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Splunk SOAR (Phantom) automation agent for playbook management."""

import json
import sys
import argparse
from datetime import datetime

try:
    import requests
    from requests.packages.urllib3.exceptions import InsecureRequestWarning
    requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)


class SplunkSOARClient:
    """Client for Splunk SOAR (Phantom) REST API."""

    def __init__(self, base_url, auth_token, verify_ssl=False):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "ph-auth-token": auth_token,
            "Content-Type": "application/json",
        })
        self.session.verify = verify_ssl

    def _get(self, endpoint, params=None):
        resp = self.session.get(f"{self.base_url}/rest{endpoint}", params=params, timeout=30)
        resp.raise_for_status()
        return resp.json()

    def _post(self, endpoint, data=None):
        resp = self.session.post(f"{self.base_url}/rest{endpoint}", json=data, timeout=30)
        resp.raise_for_status()
        return resp.json()

    def list_playbooks(self, page_size=50):
        """List all configured playbooks."""
        return self._get("/playbook", params={"page_size": page_size})

    def get_playbook(self, playbook_id):
        """Get details of a specific playbook."""
        return self._get(f"/playbook/{playbook_id}")

    def run_playbook(self, playbook_id, container_id, scope="all"):
        """Execute a playbook against a container."""
        return self._post("/playbook_run", data={
            "playbook_id": playbook_id,
            "container_id": container_id,
            "scope": scope,
        })

    def list_containers(self, label=None, status=None, page_size=50):
        """List containers (incidents/events)."""
        params = {"page_size": page_size, "sort": "id", "order": "desc"}
        if label:
            params["_filter_label"] = f'"{label}"'
        if status:
            params["_filter_status"] = f'"{status}"'
        return self._get("/container", params=params)

    def create_container(self, name, label, severity, description=""):
        """Create a new container for an incident."""
        return self._post("/container", data={
            "name": name, "label": label,
            "severity": severity, "description": description,
            "status": "new",
        })

    def add_artifact(self, container_id, name, cef_data, label="event"):
        """Add an artifact (IOC) to a container."""
        return self._post("/artifact", data={
            "container_id": container_id,
            "name": name,
            "label": label,
            "cef": cef_data,
            "severity": "medium",
        })

    def list_apps(self):
        """List installed apps (connectors)."""
        return self._get("/app")

    def list_assets(self):
        """List configured assets."""
        return self._get("/asset")

    def get_action_results(self, action_run_id):
        """Get results of an action run."""
        return self._get(f"/action_run/{action_run_id}")

    def run_action(self, action_name, app_id, asset_id, parameters, container_id):
        """Run an action via an app connector."""
        return self._post("/action_run", data={
            "action": action_name,
            "app_id": app_id,
            "asset_id": asset_id,
            "container_id": container_id,
            "parameters": [parameters],
        })

    def get_system_info(self):
        """Get SOAR system information."""
        return self._get("/system_info")

    def list_users(self):
        """List SOAR users."""
        return self._get("/ph_user")


def create_phishing_response_playbook_data():
    """Generate phishing response playbook configuration."""
    return {
        "name": "Phishing Investigation and Response",
        "description": "Automated phishing email triage and response",
        "steps": [
            {"action": "file_reputation", "app": "VirusTotal",
             "description": "Check attachment hash against VT"},
            {"action": "url_reputation", "app": "VirusTotal",
             "description": "Check URLs in email against VT"},
            {"action": "domain_reputation", "app": "VirusTotal",
             "description": "Check sender domain reputation"},
            {"action": "whois_domain", "app": "WHOIS",
             "description": "WHOIS lookup on sender domain"},
            {"action": "hunt_email", "app": "Exchange",
             "description": "Search for same email across mailboxes"},
            {"action": "decision_gate", "type": "prompt",
             "description": "Analyst reviews enrichment and decides"},
            {"action": "quarantine_email", "app": "Exchange",
             "description": "Quarantine email from all mailboxes"},
            {"action": "block_sender", "app": "Firewall",
             "description": "Block sender IP/domain on email gateway"},
            {"action": "create_ticket", "app": "ServiceNow",
             "description": "Create incident ticket for tracking"},
        ],
    }


def create_malware_containment_playbook_data():
    """Generate malware containment playbook configuration."""
    return {
        "name": "Malware Containment and Remediation",
        "steps": [
            {"action": "get_process_info", "app": "CrowdStrike",
             "description": "Get process details from EDR"},
            {"action": "file_reputation", "app": "VirusTotal",
             "description": "Check file hash reputation"},
            {"action": "detonate_file", "app": "Sandbox",
             "description": "Detonate in sandbox if unknown"},
            {"action": "decision_gate", "type": "prompt",
             "description": "Analyst approves containment"},
            {"action": "contain_device", "app": "CrowdStrike",
             "description": "Network isolate the endpoint"},
            {"action": "disable_user", "app": "ActiveDirectory",
             "description": "Disable compromised user account"},
            {"action": "create_ticket", "app": "ServiceNow",
             "description": "Create P1 incident ticket"},
        ],
    }


def run_soar_audit(client):
    """Run SOAR platform audit."""
    print(f"\n{'='*60}")
    print(f"  SPLUNK SOAR (PHANTOM) AUDIT")
    print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
    print(f"{'='*60}\n")

    try:
        sys_info = client.get_system_info()
        print(f"--- SYSTEM INFO ---")
        print(f"  Version: {sys_info.get('version', 'N/A')}")
        print(f"  Build: {sys_info.get('build', 'N/A')}")
    except Exception as e:
        print(f"  System info unavailable: {e}")

    playbooks = client.list_playbooks()
    pb_data = playbooks.get("data", [])
    print(f"\n--- PLAYBOOKS ({len(pb_data)}) ---")
    for pb in pb_data[:15]:
        status = "ACTIVE" if pb.get("active") else "INACTIVE"
        print(f"  [{status}] {pb.get('name', 'N/A')} (ID: {pb.get('id')})")

    apps = client.list_apps()
    app_data = apps.get("data", [])
    print(f"\n--- INSTALLED APPS ({len(app_data)}) ---")
    for app in app_data[:15]:
        print(f"  {app.get('name', 'N/A')} v{app.get('app_version', 'N/A')}")

    assets = client.list_assets()
    asset_data = assets.get("data", [])
    print(f"\n--- CONFIGURED ASSETS ({len(asset_data)}) ---")
    for asset in asset_data[:10]:
        print(f"  {asset.get('name', 'N/A')} -> {asset.get('product_name', 'N/A')}")

    containers = client.list_containers(status="open")
    ct_data = containers.get("data", [])
    print(f"\n--- OPEN CONTAINERS ({len(ct_data)}) ---")
    for ct in ct_data[:10]:
        print(f"  [{ct.get('severity', 'N/A')}] {ct.get('name', 'N/A')} (Status: {ct.get('status')})")

    print(f"\n--- PLAYBOOK TEMPLATES ---")
    phishing = create_phishing_response_playbook_data()
    print(f"  {phishing['name']}: {len(phishing['steps'])} steps")
    malware = create_malware_containment_playbook_data()
    print(f"  {malware['name']}: {len(malware['steps'])} steps")

    print(f"\n{'='*60}\n")
    return {"playbooks": len(pb_data), "apps": len(app_data), "containers": len(ct_data)}


def main():
    parser = argparse.ArgumentParser(description="Splunk SOAR Automation Agent")
    parser.add_argument("--url", required=True, help="SOAR instance URL")
    parser.add_argument("--token", required=True, help="SOAR auth token")
    parser.add_argument("--audit", action="store_true", help="Run SOAR audit")
    parser.add_argument("--list-playbooks", action="store_true")
    parser.add_argument("--run-playbook", nargs=2, metavar=("PB_ID", "CONTAINER_ID"),
                        help="Run playbook on container")
    parser.add_argument("--output", help="Save report to JSON")
    args = parser.parse_args()

    client = SplunkSOARClient(args.url, args.token)

    if args.audit:
        report = run_soar_audit(client)
        if args.output:
            with open(args.output, "w") as f:
                json.dump(report, f, indent=2, default=str)
    elif args.list_playbooks:
        pb = client.list_playbooks()
        for p in pb.get("data", []):
            print(f"  [{p.get('id')}] {p.get('name')}")
    elif args.run_playbook:
        result = client.run_playbook(int(args.run_playbook[0]), int(args.run_playbook[1]))
        print(json.dumps(result, indent=2))
    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Keep exploring