npx skills add mukul975/Anthropic-Cybersecurity-SkillsMITRE ATT&CK
NIST CSF 2.0
MITRE D3FEND
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, ServiceFileNameStep 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.
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 clientget_groups()-- Retrieve all threat actor groups from ATT&CKget_techniques_used_by_group(group)-- Get techniques mapped to a specific groupget_techniques()-- List all ATT&CK techniquesget_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 bundleget_groups()-- All threat groupsget_techniques_used_by_group(group_stix_id)-- Techniques per groupget_attack_campaigns()-- Known campaigns
osquery
- Docs: https://osquery.readthedocs.io/
scheduled_tasks-- Windows scheduled tasks tableprocesses-- Running process informationprocess_open_sockets-- Network connections per processautoexec-- Auto-start execution pointsfile-- 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
- Spec: https://sigmahq.io/docs/basics/rules.html
- Fields:
title,status,logsource,detection,level - Converters:
sigma-cliconverts to Splunk SPL, Elastic EQL, Sentinel KQL
External References
- MITRE ATT&CK Groups: https://attack.mitre.org/groups/
- ATT&CK Navigator: https://mitre-attack.github.io/attack-navigator/
- Velociraptor VQL: https://docs.velociraptor.app/docs/vql/
- Zeek Documentation: https://docs.zeek.org/en/current/
Scripts 1
agent.py7.1 KB
#!/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()