threat intelligence

Hunting Advanced Persistent Threats

Proactively hunts for Advanced Persistent Threat (APT) activity within enterprise environments using hypothesis-driven searches across endpoint telemetry, network logs, and memory artifacts. Use when conducting scheduled threat hunting cycles, investigating anomalous behavior flagged by UEBA, or validating that known APT TTPs are not present in the environment. Activates for requests involving MITRE ATT&CK, Velociraptor, osquery, Zeek, or threat hunting playbooks.

aptedrmitre-att&cknist-csfosquerythreat-huntingttpvelociraptor
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

Use this skill when:

  • Conducting proactive threat hunting sprints (typically 2–4 week cycles) based on newly published APT intelligence
  • A UEBA alert or anomaly detection system flags behavioral deviations warranting deeper investigation
  • A peer organization or ISAC sharing partner reports active APT compromise and you need to validate your own exposure

Do not use this skill as a substitute for incident response when a confirmed breach is in progress — escalate to IR procedures (NIST SP 800-61).

Prerequisites

  • EDR platform with telemetry retention (CrowdStrike Falcon, Microsoft Defender for Endpoint, or SentinelOne) covering 30+ days
  • Access to MITRE ATT&CK Navigator for hypothesis development
  • Network flow data (NetFlow, Zeek, or Suricata logs) in a queryable SIEM
  • Threat hunting platform or query interface (Velociraptor, osquery fleet, or Splunk ES)

Workflow

Step 1: Develop Hunt Hypothesis

Select a threat actor relevant to your sector using MITRE ATT&CK Groups (https://attack.mitre.org/groups/). Review the group's known TTPs mapped to ATT&CK techniques. Example hypothesis: "APT29 (Cozy Bear) uses spearphishing with ISO attachments (T1566.001) and living-off-the-land binaries (T1218) — test for unusual mshta.exe and rundll32.exe parent-child relationships."

Document hypothesis using the Threat Hunting Loop framework: hypothesis → data collection → pattern analysis → response.

Step 2: Identify Required Data Sources

Map each ATT&CK technique to required log sources using the ATT&CK Data Sources taxonomy:

  • Process creation (T1059): Windows Security Event 4688 or Sysmon Event ID 1
  • Network connections (T1071): Zeek conn.log, NetFlow, EDR network telemetry
  • Registry modifications (T1547): Sysmon Event ID 13, Windows Security 4657
  • Memory injection (T1055): EDR memory scan telemetry, Volatility output

Verify log coverage using ATT&CK Coverage Calculator or a custom data source matrix.

Step 3: Execute Hunts with Velociraptor or osquery

Velociraptor VQL hunt for unusual PowerShell execution:

SELECT Pid, Ppid, Name, CommandLine, CreateTime
FROM pslist()
WHERE Name =~ "powershell.exe"
AND CommandLine =~ "-enc|-nop|-w hidden"

osquery for persistence via scheduled tasks:

SELECT name, action, enabled, path
FROM scheduled_tasks
WHERE action NOT LIKE '%System32%'
AND enabled = 1;

Splunk SPL for lateral movement via PsExec:

index=windows EventCode=7045 ServiceFileName="*PSEXESVC*"
| stats count by ComputerName, ServiceName, ServiceFileName

Step 4: Analyze Results and Pivot

For each anomaly identified, pivot across dimensions:

  • Temporal: Did this occur before or after known IOC timestamps?
  • Host: How many endpoints exhibit this behavior?
  • User: Is the associated account a service account, privileged user, or regular user?
  • Network: Does the host communicate with external IPs not in baseline?

Apply the Diamond Model (adversary, capability, infrastructure, victim) to structure findings.

Step 5: Document and Operationalize Findings

If hunting reveals confirmed malicious activity, activate IR procedures. If hunting reveals a gap (hunt found nothing but data coverage was insufficient), document the coverage gap and remediate.

Convert successful hunt queries into SIEM detection rules using Sigma format for portability across platforms.

Key Concepts

Term Definition
TTP Tactics, Techniques, and Procedures — adversary behavioral patterns as defined in MITRE ATT&CK
Diamond Model Analytical framework with four vertices (adversary, capability, infrastructure, victim) used to structure intrusion analysis
Living-off-the-Land (LotL) Attacker technique using legitimate OS tools (PowerShell, WMI, certutil) to evade detection
UEBA User and Entity Behavior Analytics — ML-based detection of anomalous behavior baselines
Sigma Open standard for SIEM-agnostic detection rule format, analogous to YARA for network/log detection
Hunt Hypothesis A testable prediction about adversary presence based on threat intelligence and environmental knowledge

Tools & Systems

  • Velociraptor: Open-source DFIR platform with VQL query language for scalable endpoint hunting across thousands of systems
  • osquery: SQL-based OS instrumentation framework for real-time endpoint telemetry queries
  • MITRE ATT&CK Navigator: Web-based tool for visualizing ATT&CK coverage and technique prioritization
  • Zeek (formerly Bro): Network traffic analyzer producing structured logs (conn, dns, http, ssl) suitable for hunting
  • Elastic Security: EQL (Event Query Language) enables sequence-based hunting for multi-stage attack patterns
  • Sigma: Detection rule format with translators for Splunk, QRadar, Sentinel, and Elastic

Common Pitfalls

  • Confirmation bias: Starting a hunt expecting to find something and interpreting benign data as malicious. Document null results — they validate controls.
  • Insufficient data retention: Many APT techniques require 90+ days of log history to identify slow-and-low patterns. Default retention periods are often too short.
  • Hunting without baselines: Cannot identify anomalies without knowing normal. Spend time on baseline documentation before hunting.
  • Query performance impact: Broad queries against production SIEM during business hours can degrade analyst workflows. Schedule intensive hunts during off-peak hours.
  • Ignoring false positives systematically: Track false positive rates per query. Queries with >80% FP rate should be refined or retired before operationalization.
Source materials

References and resources

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

References 1

api-reference.md2.1 KB

API Reference: Hunting Advanced Persistent Threats

Libraries

attackcti (MITRE ATT&CK CTI Library)

  • Install: pip install attackcti
  • Docs: https://attackcti.readthedocs.io/
  • attack_client() -- Initialize the ATT&CK STIX/TAXII client
  • get_groups() -- Retrieve all threat actor groups from ATT&CK
  • get_techniques_used_by_group(group) -- Get techniques mapped to a specific group
  • get_techniques() -- List all ATT&CK techniques
  • get_mitigations() -- List all mitigations

mitreattack-python (ATT&CK STIX Data)

  • Install: pip install mitreattack-python
  • Docs: https://mitreattack-python.readthedocs.io/
  • MitreAttackData(stix_filepath) -- Load ATT&CK STIX bundle
  • get_groups() -- All threat groups
  • get_techniques_used_by_group(group_stix_id) -- Techniques per group
  • get_attack_campaigns() -- Known campaigns

osquery

  • Docs: https://osquery.readthedocs.io/
  • scheduled_tasks -- Windows scheduled tasks table
  • processes -- Running process information
  • process_open_sockets -- Network connections per process
  • autoexec -- Auto-start execution points
  • file -- File metadata queries

Key ATT&CK Technique IDs

ID Name Relevance
T1059 Command and Scripting Interpreter Process-based hunting
T1053 Scheduled Task/Job Persistence detection
T1071 Application Layer Protocol C2 communication
T1055 Process Injection In-memory threats
T1003 OS Credential Dumping Credential theft
T1566 Phishing Initial access vector
T1218 Signed Binary Proxy Execution Defense evasion

Sigma Rule Format

External References

Scripts 1

agent.py7.1 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""APT threat hunting agent using MITRE ATT&CK, attackcti, and osquery."""

import json
import sys
import argparse
from datetime import datetime

try:
    from attackcti import attack_client
except ImportError:
    print("Install attackcti: pip install attackcti")
    sys.exit(1)


def get_apt_group_ttps(group_name):
    """Retrieve TTPs for a specific APT group from MITRE ATT&CK."""
    client = attack_client()
    groups = client.get_groups()
    target = None
    for g in groups:
        aliases = [a.lower() for a in g.get("aliases", [])]
        if group_name.lower() in g["name"].lower() or group_name.lower() in aliases:
            target = g
            break
    if not target:
        print(f"[!] Group '{group_name}' not found in ATT&CK")
        return None
    techniques = client.get_techniques_used_by_group(target)
    return {"group": target["name"], "id": target["external_references"][0]["external_id"],
            "techniques": [{"id": t["external_references"][0]["external_id"],
                            "name": t["name"],
                            "tactic": [p["phase_name"] for p in t.get("kill_chain_phases", [])]}
                           for t in techniques]}


def generate_osquery_hunts(techniques):
    """Generate osquery hunt queries for detected ATT&CK techniques."""
    query_map = {
        "T1059": ("Process execution (Command and Scripting)",
                  "SELECT pid, name, cmdline, path, parent FROM processes "
                  "WHERE name IN ('powershell.exe','cmd.exe','wscript.exe','cscript.exe','bash','python');"),
        "T1053": ("Scheduled Task/Job persistence",
                  "SELECT name, action, path, enabled, last_run_time FROM scheduled_tasks "
                  "WHERE enabled=1 AND action NOT LIKE '%System32%';"),
        "T1547": ("Boot/Logon autostart execution",
                  "SELECT name, path, source FROM autoexec;"),
        "T1071": ("Application layer protocol C2",
                  "SELECT pid, remote_address, remote_port, local_port FROM process_open_sockets "
                  "WHERE remote_port IN (443, 8443, 8080, 4443) AND family=2;"),
        "T1055": ("Process injection",
                  "SELECT pid, name, cmdline FROM processes WHERE on_disk=0;"),
        "T1003": ("OS credential dumping",
                  "SELECT pid, name, cmdline FROM processes "
                  "WHERE name IN ('mimikatz.exe','procdump.exe','ntdsutil.exe') "
                  "OR cmdline LIKE '%sekurlsa%' OR cmdline LIKE '%lsass%';"),
        "T1021": ("Remote services lateral movement",
                  "SELECT pid, name, cmdline FROM processes "
                  "WHERE name IN ('psexec.exe','wmic.exe','winrm.cmd') "
                  "OR cmdline LIKE '%invoke-command%';"),
        "T1027": ("Obfuscated files or information",
                  "SELECT pid, name, cmdline FROM processes "
                  "WHERE cmdline LIKE '%-enc%' OR cmdline LIKE '%-encodedcommand%';"),
        "T1566": ("Phishing initial access",
                  "SELECT path, filename, size FROM file "
                  "WHERE directory LIKE '%Downloads%' "
                  "AND (filename LIKE '%.iso' OR filename LIKE '%.img' OR filename LIKE '%.lnk');"),
        "T1218": ("Signed binary proxy execution",
                  "SELECT pid, name, cmdline, parent FROM processes "
                  "WHERE name IN ('mshta.exe','rundll32.exe','regsvr32.exe','certutil.exe');"),
    }
    hunts = []
    for tech in techniques:
        tech_id = tech["id"].split(".")[0]
        if tech_id in query_map:
            desc, query = query_map[tech_id]
            hunts.append({"technique": tech["id"], "name": tech["name"],
                          "description": desc, "osquery": query})
    return hunts


def generate_sigma_rule(technique_id, technique_name, tactic):
    """Generate a Sigma detection rule for a given technique."""
    return {
        "title": f"Detect {technique_name} ({technique_id})",
        "status": "experimental",
        "description": f"Detects potential {technique_name} activity mapped to {technique_id}",
        "references": [f"https://attack.mitre.org/techniques/{technique_id.replace('.','/')}/"],
        "tags": [f"attack.{t}" for t in tactic] + [f"attack.{technique_id.lower()}"],
        "logsource": {"category": "process_creation", "product": "windows"},
        "detection": {"selection": {"technique_id": technique_id}, "condition": "selection"},
        "level": "medium",
    }


def build_hunt_report(group_name):
    """Build a complete threat hunt report for an APT group."""
    print(f"\n{'='*70}")
    print(f"  APT THREAT HUNT REPORT")
    print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
    print(f"{'='*70}\n")

    print(f"[*] Querying MITRE ATT&CK for group: {group_name}")
    group_data = get_apt_group_ttps(group_name)
    if not group_data:
        return

    print(f"[+] Found: {group_data['group']} ({group_data['id']})")
    print(f"[+] Techniques mapped: {len(group_data['techniques'])}\n")

    print(f"--- TECHNIQUE COVERAGE ---")
    tactic_counts = {}
    for t in group_data["techniques"]:
        print(f"  [{t['id']}] {t['name']} -> {', '.join(t['tactic'])}")
        for tac in t["tactic"]:
            tactic_counts[tac] = tactic_counts.get(tac, 0) + 1

    print(f"\n--- TACTIC DISTRIBUTION ---")
    for tac, count in sorted(tactic_counts.items(), key=lambda x: -x[1]):
        bar = "#" * count
        print(f"  {tac:<30} {bar} ({count})")

    print(f"\n--- OSQUERY HUNT QUERIES ---")
    hunts = generate_osquery_hunts(group_data["techniques"])
    if hunts:
        for h in hunts:
            print(f"\n  Technique: {h['technique']} - {h['description']}")
            print(f"  Query: {h['osquery']}")
    else:
        print("  No matching osquery hunts for this group's techniques.")

    print(f"\n--- SIGMA RULES ---")
    for t in group_data["techniques"][:5]:
        rule = generate_sigma_rule(t["id"], t["name"], t["tactic"])
        print(f"\n  Rule: {rule['title']}")
        print(f"  Tags: {', '.join(rule['tags'])}")
        print(f"  Level: {rule['level']}")

    print(f"\n--- HUNT RECOMMENDATIONS ---")
    print(f"  1. Execute osquery hunts across all endpoints via fleet manager")
    print(f"  2. Search SIEM for technique indicators over past 90 days")
    print(f"  3. Validate EDR telemetry covers all {len(group_data['techniques'])} techniques")
    print(f"  4. Cross-reference with network logs (Zeek/Suricata) for C2 patterns")
    print(f"  5. Document findings using Diamond Model analysis framework")
    print(f"\n{'='*70}\n")

    return group_data


def main():
    parser = argparse.ArgumentParser(description="APT Threat Hunting Agent")
    parser.add_argument("--group", default="APT29", help="APT group name (e.g., APT29, APT28, Lazarus)")
    parser.add_argument("--output", help="Save report to JSON file")
    args = parser.parse_args()

    report = build_hunt_report(args.group)
    if report and args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"[+] JSON report saved to {args.output}")


if __name__ == "__main__":
    main()
Keep exploring