
Threat Detection
- 605 installs
- 23.5k repo stars
- Updated July 17, 2026
- alirezarezvani/claude-skills
threat-detection is a Claude Code security skill that hunts threats, analyzes IOCs, and detects behavioral anomalies in production telemetry for developers and security engineers who need proactive discovery beyond autom
About
threat-detection is a security operations skill from alirezarezvani/claude-skills focused on proactive threat discovery—not incident response or red-team exercises. The skill covers hypothesis-driven threat hunting, IOC sweep generation, z-score behavioral anomaly detection, and MITRE ATT&CK-mapped signal prioritization across production logs and telemetry. Security and platform engineers reach for threat-detection when automated controls may have missed lateral movement, C2 beacons, or subtle credential abuse. Outputs include prioritized hunt hypotheses, IOC query batches, and anomaly-ranked signals mapped to ATT&CK techniques for further investigation.
- Hypothesis-driven threat hunting methodology
- IOC sweep generation and analysis
- Z-score based anomaly detection on telemetry
- MITRE ATT&CK-mapped signal prioritization
- Deception and honeypot integration patterns
Threat Detection by the numbers
- 605 all-time installs (skills.sh)
- Ranked #478 of 2,203 Security skills by installs in the Skillselion catalog
- Security screen: HIGH risk (skills.sh audit)
- Data as of Jul 31, 2026 (Skillselion catalog sync)
npx skills add https://github.com/alirezarezvani/claude-skills --skill threat-detectionAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 605 |
|---|---|
| repo stars | ★ 23.5k |
| Security audit | 2 / 3 scanners passed |
| Last updated | July 17, 2026 |
| Repository | alirezarezvani/claude-skills ↗ |
How do you hunt threats in production telemetry?
Proactively hunt for threats, analyze IOCs, and detect behavioral anomalies in their production systems and telemetry.
Who is it for?
Security and platform engineers with production log access who need structured hunts for IOCs and behavioral anomalies evading automated controls.
Skip if: Developers needing incident-response playbooks, penetration-test exploitation steps, or compliance policy authoring without telemetry data.
When should I use this skill?
A developer or security engineer asks to hunt threats, analyze IOCs, detect telemetry anomalies, or prioritize signals using MITRE ATT&CK.
What you get
Threat hunt hypotheses, IOC sweep queries, z-score anomaly reports, and MITRE ATT&CK-prioritized signal lists.
- IOC sweep queries
- prioritized threat signals
Files
Threat Detection
Threat detection skill for proactive discovery of attacker activity through hypothesis-driven hunting, IOC analysis, and behavioral anomaly detection. This is NOT incident response (see incident-response) or red team operations (see red-team) — this is about finding threats that have evaded automated controls.
---
Table of Contents
- Overview
- Threat Signal Analyzer
- Threat Hunting Methodology
- IOC Analysis
- Anomaly Detection
- MITRE ATT&CK Signal Prioritization
- Deception and Honeypot Integration
- Workflows
- Anti-Patterns
- Cross-References
---
Overview
What This Skill Does
This skill provides the methodology and tooling for proactive threat detection — finding attacker activity through structured hunting hypotheses, IOC analysis, and statistical anomaly detection before alerts fire.
Distinction from Other Security Skills
| Skill | Focus | Approach |
|---|---|---|
| threat-detection (this) | Finding hidden threats | Proactive — hunt before alerts |
| incident-response | Active incidents | Reactive — contain and investigate declared incidents |
| red-team | Offensive simulation | Offensive — test defenses from attacker perspective |
| cloud-security | Cloud misconfigurations | Posture — IAM, S3, network exposure |
Prerequisites
Read access to SIEM/EDR telemetry, endpoint logs, and network flow data. IOC feeds require freshness within 30 days to avoid false positives. Hunting hypotheses must be scoped to the environment before execution.
---
Threat Signal Analyzer
The threat_signal_analyzer.py tool supports three modes: hunt (hypothesis scoring), ioc (sweep generation), and anomaly (statistical detection).
# Hunt mode: score a hypothesis against MITRE ATT&CK coverage
python3 scripts/threat_signal_analyzer.py --mode hunt \
--hypothesis "Lateral movement via PtH using compromised service account" \
--actor-relevance 3 --control-gap 2 --data-availability 2 --json
# IOC mode: generate sweep targets from an IOC feed file
python3 scripts/threat_signal_analyzer.py --mode ioc \
--ioc-file iocs.json --json
# Anomaly mode: detect statistical outliers in telemetry events
python3 scripts/threat_signal_analyzer.py --mode anomaly \
--events-file telemetry.json \
--baseline-mean 100 --baseline-std 25 --json
# List all supported MITRE ATT&CK techniques
python3 scripts/threat_signal_analyzer.py --list-techniquesIOC file format
{
"ips": ["1.2.3.4", "5.6.7.8"],
"domains": ["malicious.example.com"],
"hashes": ["abc123def456..."]
}Telemetry events file format
[
{"timestamp": "2024-01-15T14:32:00Z", "entity": "host-01", "action": "dns_query", "volume": 450},
{"timestamp": "2024-01-15T14:33:00Z", "entity": "host-02", "action": "dns_query", "volume": 95}
]Exit codes
| Code | Meaning |
|---|---|
| 0 | No high-priority findings |
| 1 | Medium-priority signals detected |
| 2 | High-priority confirmed findings |
---
Threat Hunting Methodology
Structured threat hunting follows a five-step loop: hypothesis → data source identification → query execution → finding triage → feedback to detection engineering.
Hypothesis Scoring
| Factor | Weight | Description |
|---|---|---|
| Actor relevance | ×3 | How closely does this TTP match known threat actors in your sector? |
| Control gap | ×2 | How many of your existing controls would miss this behavior? |
| Data availability | ×1 | Do you have the telemetry data needed to test this hypothesis? |
Priority score = (actor_relevance × 3) + (control_gap × 2) + (data_availability × 1)
High-Value Hunt Hypotheses by Tactic
| Hypothesis | MITRE ID | Data Sources | Priority Signal |
|---|---|---|---|
| WMI lateral movement via remote execution | T1047 | WMI logs, EDR process telemetry | WMI process spawned from WINRM, unusual parent-child chain |
| LOLBin execution for defense evasion | T1218 | Process creation, command-line args | certutil.exe, regsvr32.exe, mshta.exe with network activity |
| Beaconing C2 via jitter-heavy intervals | T1071.001 | Proxy logs, DNS logs | Regular interval outbound connections ±10% jitter |
| Pass-the-Hash lateral movement | T1550.002 | Windows security event 4624 type 3 | NTLM auth from unexpected source host to admin share |
| LSASS memory access | T1003.001 | EDR memory access events | OpenProcess on lsass.exe from non-system process |
| Kerberoasting | T1558.003 | Windows event 4769 | High volume TGS requests for service accounts |
| Scheduled task persistence | T1053.005 | Sysmon Event 1/11, Windows 4698 | Scheduled task created in non-standard directory |
---
IOC Analysis
IOC analysis determines whether indicators are fresh, maps them to required sweep targets, and filters stale data that generates false positives.
IOC Types and Sweep Priority
| IOC Type | Staleness Threshold | Sweep Target | MITRE Coverage |
|---|---|---|---|
| IP addresses | 30 days | Firewall logs, NetFlow, proxy logs | T1071, T1105 |
| Domains | 30 days | DNS resolver logs, proxy logs | T1568, T1583 |
| File hashes | 90 days | EDR file creation, AV scan logs | T1105, T1027 |
| URLs | 14 days | Proxy access logs, browser history | T1566.002 |
| Mutex names | 180 days | EDR runtime artifacts | T1055 |
IOC Staleness Handling
IOCs older than their threshold are flagged as stale and excluded from sweep target generation. Running sweeps against stale IOCs inflates false positive rates and reduces SOC credibility. Refresh IOC feeds from threat intelligence platforms (MISP, OpenCTI, commercial TI) before every hunt cycle.
---
Anomaly Detection
Statistical anomaly detection identifies behavior that deviates from established baselines without relying on known-bad signatures.
Z-Score Thresholds
| Z-Score | Classification | Response |
|---|---|---|
| < 2.0 | Normal | No action required |
| 2.0–2.9 | Soft anomaly | Log and monitor — increase sampling |
| ≥ 3.0 | Hard anomaly | Escalate to hunt analyst — investigate entity |
Baseline Requirements
Effective anomaly detection requires at least 14 days of historical telemetry to establish a valid baseline. Baselines must be recomputed after:
- Security incidents (post-incident behavior change)
- Major infrastructure changes (cloud migrations, new SaaS deployments)
- Seasonal usage pattern changes (end of quarter, holiday periods)
High-Value Anomaly Targets
| Entity Type | Metric | Anomaly Indicator |
|---|---|---|
| DNS resolver | Queries per hour per host | Beaconing, tunneling, DGA |
| Endpoint | Unique process executions per day | Malware installation, LOLBin abuse |
| Service account | Auth events per hour | Credential stuffing, lateral movement |
| Email gateway | Attachment types per hour | Phishing campaign spike |
| Cloud IAM | API calls per identity per hour | Credential compromise, exfiltration |
---
MITRE ATT&CK Signal Prioritization
Each hunting hypothesis maps to one or more ATT&CK techniques. Techniques with multiple confirmed signals in your environment are higher priority.
Tactic Coverage Matrix
| Tactic | Key Techniques | Primary Data Source | |--------|---------------|--------------------|-| | Initial Access | T1190, T1566, T1078 | Web access logs, email gateway, auth logs | | Execution | T1059, T1047, T1218 | Process creation, command-line, script execution | | Persistence | T1053, T1543, T1098 | Scheduled tasks, services, account changes | | Defense Evasion | T1027, T1562, T1070 | Process hollowing, log clearing, encoding | | Credential Access | T1003, T1558, T1110 | LSASS, Kerberos, auth failures | | Lateral Movement | T1550, T1021, T1534 | NTLM auth, remote services, internal spearphish | | Collection | T1074, T1560, T1114 | Staging directories, archive creation, email access | | Exfiltration | T1048, T1041, T1567 | Unusual outbound volume, DNS tunneling, cloud storage | | Command & Control | T1071, T1572, T1568 | Beaconing, protocol tunneling, DNS C2 |
---
Deception and Honeypot Integration
Deception assets generate high-fidelity alerts — any interaction with a honeypot is an unambiguous signal requiring investigation.
Deception Asset Types and Placement
| Asset Type | Placement | Signal | ATT&CK Technique |
|---|---|---|---|
| Honeypot credentials in password vault | Vault secrets store | Credential access attempt | T1555 |
| Honey tokens (fake AWS access keys) | Git repos, S3 objects | Reconnaissance or exfiltration | T1552.004 |
| Honey files (named: passwords.xlsx) | File shares, endpoints | Collection staging | T1074 |
| Honey accounts (dormant AD users) | Active Directory | Lateral movement pivot | T1078.002 |
| Honeypot network services | DMZ, flat network segments | Network scanning, service exploitation | T1046, T1190 |
Honeypot alerts bypass the standard scoring pipeline — any hit is an automatic SEV2 until proven otherwise.
---
Workflows
Workflow 1: Quick Hunt (30 Minutes)
For responding to a new threat intelligence report or CVE alert:
# 1. Score hypothesis against environment context
python3 scripts/threat_signal_analyzer.py --mode hunt \
--hypothesis "Exploitation of CVE-YYYY-NNNNN in Apache" \
--actor-relevance 2 --control-gap 3 --data-availability 2 --json
# 2. Build IOC sweep list from threat intel
echo '{"ips": ["1.2.3.4"], "domains": ["malicious.tld"], "hashes": []}' > iocs.json
python3 scripts/threat_signal_analyzer.py --mode ioc --ioc-file iocs.json --json
# 3. Check for anomalies in web server telemetry from last 24h
python3 scripts/threat_signal_analyzer.py --mode anomaly \
--events-file web_events_24h.json --baseline-mean 80 --baseline-std 20 --jsonDecision: If hunt priority ≥ 7 or any IOC sweep hits, escalate to full hunt.
Workflow 2: Full Threat Hunt (Multi-Day)
Day 1 — Hypothesis Generation: 1. Review threat intelligence feeds for sector-relevant TTPs 2. Map last 30 days of security alerts to ATT&CK tactics to identify gaps 3. Score top 5 hypotheses with threat_signal_analyzer.py hunt mode 4. Prioritize by score — start with highest
Day 2 — Data Collection and Query Execution: 1. Pull relevant telemetry from SIEM (date range: last 14 days) 2. Run anomaly detection across entity baselines 3. Execute IOC sweeps for all feeds fresh within 30 days 4. Review hunt playbooks in references/hunt-playbooks.md
Day 3 — Triage and Reporting: 1. Triage all anomaly findings — confirm or dismiss 2. Escalate confirmed activity to incident-response 3. Document new detection rules from hunt findings 4. Submit false-positive IOCs back to TI provider
Workflow 3: Continuous Monitoring (Automated)
Configure recurring anomaly detection against key entity baselines on a 6-hour cadence:
# Run as cron job every 6 hours — auto-escalate on exit code 2
python3 scripts/threat_signal_analyzer.py --mode anomaly \
--events-file /var/log/telemetry/events_6h.json \
--baseline-mean "${BASELINE_MEAN}" \
--baseline-std "${BASELINE_STD}" \
--json > /var/log/threat-detection/$(date +%Y%m%d_%H%M%S).json
# Alert on exit code 2 (hard anomaly)
if [ $? -eq 2 ]; then
send_alert "Hard anomaly detected — threat_signal_analyzer"
fi---
Anti-Patterns
1. Hunting without a hypothesis — Running broad queries across all telemetry without a focused question generates noise, not signal. Every hunt must start with a testable hypothesis scoped to one or two ATT&CK techniques. 2. Using stale IOCs — IOCs older than 30 days generate false positives that train analysts to ignore alerts. Always check IOC freshness before sweeping; exclude stale indicators from automated sweeps. 3. Skipping baseline establishment — Anomaly detection without a valid baseline produces alerts on normal high-volume days. Require 14+ days of baseline data before enabling statistical alerting on any entity type. 4. Hunting only known techniques — Hunting exclusively against documented ATT&CK techniques misses novel adversary behavior. Regularly include open-ended anomaly analysis that can surface unknown TTPs. 5. Not closing the feedback loop to detection engineering — Hunt findings that confirm malicious behavior must produce new detection rules. Hunting that doesn't improve detection coverage has no lasting value. 6. Treating every anomaly as a confirmed threat — High z-scores indicate deviation from baseline, not confirmed malice. All anomalies require human triage to confirm or dismiss before escalation. 7. Ignoring honeypot alerts — Any interaction with a deception asset is a high-fidelity signal. Treating honeypot alerts as noise invalidates the entire deception investment.
---
Cross-References
| Skill | Relationship |
|---|---|
| incident-response | Confirmed threats from hunting escalate to incident-response for triage and containment |
| red-team | Red team exercises generate realistic TTPs that inform hunt hypothesis prioritization |
| cloud-security | Cloud posture findings (open S3, IAM wildcards) create hunting targets for data exfiltration TTPs |
| security-pen-testing | Pen test findings identify attack surfaces that threat hunting should monitor post-remediation |
Threat Hunt Playbooks
Defensive documentation — not malware. This file lists detection queries
and indicators-of-attack for blue-team threat hunting. It cites legitimate
Windows binaries (certutil.exe,regsvr32.exe,mshta.exe,msiexec.exe,
rundll32.exe) and the LOLBin command-line patterns associated with theirabuse. No executable code is shipped here.
>
Some endpoint AV/EDR products (Bitdefender, Defender, etc.) heuristically
flag plain-text documents that contain these strings. If your scanner
quarantines this file, allow-list the path
engineering-team/skills/threat-detection/references/hunt-playbooks.mdor exclude the claude-skills checkout. The strings appear inside markdowncode spans / tables; they cannot execute from a .md file. Tracking issue:#533.
Reference playbooks for common high-value hunt hypotheses. Each playbook defines the hypothesis, required data sources, query approach, and confirmation criteria.
---
Playbook 1: WMI-Based Lateral Movement
Hypothesis: An attacker is using Windows Management Instrumentation (WMI) for remote code execution as part of lateral movement.
MITRE Technique: T1047 — Windows Management Instrumentation
Data Sources Required:
- WMI activity logs (Microsoft-Windows-WMI-Activity/Operational)
- Sysmon Event ID 1 (Process Create) and Event ID 20 (WmiEvent)
- EDR process telemetry
Query Approach: 1. Search for WMI processes (WmiPrvSE.exe, scrcons.exe) spawning child processes other than WmiApSrv.exe 2. Filter for WMI events where ActiveScriptEventConsumer or CommandLineEventConsumer is created 3. Cross-reference source host with authentication logs for lateral movement source identification
Confirmation Criteria:
- WMI child process execution on a host where the triggering identity is not the local admin or system
- WMI execution targeting multiple hosts within a short time window (>3 hosts in 10 minutes = high confidence)
False Positive Sources:
- SCCM/Configuration Manager uses WMI heavily for inventory — whitelist SCCM service accounts
- Monitoring agents (SolarWinds, Nagios) use WMI for performance data — whitelist monitoring identities
---
Playbook 2: Living-off-the-Land Binary (LOLBin) Execution
Hypothesis: An attacker is using legitimate Windows binaries (certutil.exe, regsvr32.exe, mshta.exe, msiexec.exe) for payload delivery or execution, bypassing application allowlisting.
MITRE Technique: T1218 — System Binary Proxy Execution
Data Sources Required:
- Process creation logs with full command-line (Sysmon Event ID 1)
- Network connection logs (Sysmon Event ID 3)
- DNS query logs
High-Value LOLBin Indicators:
| Binary | Suspicious Indicators | Common Abuse |
|---|---|---|
| certutil.exe | -decode or -urlcache -split -f http:// | Base64 decode, remote file download |
| regsvr32.exe | /s /u /i:http:// or scrobj.dll | Remote scriptlet execution (Squiblydoo) |
| mshta.exe | Any URL as argument | Remote HTA execution |
| msiexec.exe | /quiet /i http:// | Remote MSI execution |
| wscript.exe | Executing from temp/download directories | VBScript malware execution |
| cscript.exe | Executing from temp/download directories | JScript/VBScript malware |
| rundll32.exe | Calling exports from temp-directory DLLs | DLL side-loading |
Query Approach: 1. Search for listed LOLBins with network-connectivity-indicating arguments (URLs, IP addresses) 2. Identify LOLBin executions where the parent process is unusual (Office apps, browsers, scripting engines) 3. Flag executions from non-standard paths (temp directories, user AppData)
Confirmation Criteria:
- LOLBin making outbound network connection (Sysmon Event ID 3 within 30 seconds of Event ID 1)
- LOLBin executing from a temp or user-writable directory
- LOLBin spawned from Office application or browser process
---
Playbook 3: C2 Beaconing Detection
Hypothesis: A compromised host is communicating with a command-and-control server on a regular interval, indicating active malware or attacker control.
MITRE Technique: T1071.001 — Application Layer Protocol: Web Protocols
Data Sources Required:
- Proxy or web gateway logs (URL, user-agent, bytes transferred, connection duration)
- NetFlow or firewall session logs
- DNS resolver logs
Beaconing Indicators:
| Indicator | Threshold | Notes |
|---|---|---|
| Regular connection interval | ±10% jitter from mean | Calculate standard deviation of inter-connection times |
| Low data volume per connection | <1 KB per session | C2 check-in packets are typically small |
| Consistent user-agent string | Same UA across all requests | Hardcoded user agents in malware |
| Domain generation algorithm (DGA) | High entropy domain names | Compare against entropy baseline for org |
| Long-lived connections with low data transfer | >1 hour session, <10 KB total | HTTP long-polling C2 |
Query Approach: 1. Group outbound connections by source host + destination IP/domain 2. Calculate standard deviation of connection intervals per group 3. Flag groups where standard deviation is <10% of mean interval (regular beaconing) 4. Cross-reference destination IPs/domains against threat intel feeds
Confirmation Criteria:
- Connection regularity (coefficient of variation <0.10) from a non-browser process
- Destination domain resolves to IP with no PTR record or recently registered domain
- Connection volume inconsistent with claimed user-agent (browser UA but non-browser process)
---
Playbook 4: Pass-the-Hash Lateral Movement
Hypothesis: An attacker is using stolen NTLM hashes for lateral movement without cracking the underlying password.
MITRE Technique: T1550.002 — Use Alternate Authentication Material: Pass the Hash
Data Sources Required:
- Windows Security Event Logs (Event ID 4624 — Logon)
- Domain controller authentication logs
- EDR telemetry for LSASS memory access (pre-harvest detection)
Pass-the-Hash Indicators:
| Event | Field | Suspicious Value |
|---|---|---|
| Event 4624 | Logon Type | 3 (Network) |
| Event 4624 | Authentication Package | NTLM |
| Event 4624 | Key Length | 0 (NTLMv2) |
| Event 4624 | Source Network Address | Different from last successful logon of same account |
Query Approach: 1. Filter Event 4624 for LogonType=3 with NTLM authentication 2. Group by account name — flag accounts with authentication events from multiple source IPs within a 1-hour window 3. Correlate source hosts: the harvesting host (LSASS access) and the destination hosts (lateral movement targets) should form a pattern 4. Look for service account authentication to interactive desktop sessions (a service account logging on Type 2/10 is anomalous)
Confirmation Criteria:
- Same account authenticating to 3+ hosts via NTLM within 30 minutes
- Source hosts are workstations, not servers (server-to-server NTLM is more common legitimately)
- Account's normal authentication pattern is Kerberos — NTLM is anomalous for this identity
#!/usr/bin/env python3
"""
threat_signal_analyzer.py — Threat Signal Analysis: Hunt, IOC Sweep, Anomaly Detection
Supports three analysis modes:
hunt — Score and prioritize a threat hunting hypothesis
ioc — Process IOC list and emit sweep targets with freshness check
anomaly — Z-score behavioral anomaly detection against a baseline
Usage:
python3 threat_signal_analyzer.py --mode hunt --hypothesis "APT using WMI for lateral movement" --json
python3 threat_signal_analyzer.py --mode ioc --ioc-file iocs.json --json
python3 threat_signal_analyzer.py --mode anomaly --events-file events.json --baseline-mean 45.0 --baseline-std 12.0 --json
Exit codes:
0 No high-priority findings
1 Medium-priority signals detected
2 High-priority findings confirmed
"""
import argparse
import json
import re
import sys
from datetime import datetime, timezone
MITRE_PATTERN = r'T\d{4}(?:\.\d{3})?'
HUNT_DATA_SOURCES = {
"initial_access": ["web_proxy_logs", "email_gateway_logs", "firewall_logs", "dns_logs"],
"execution": ["edr_process_logs", "sysmon_event_1", "windows_event_4688", "auditd"],
"persistence": ["windows_event_4698", "registry_logs", "cron_logs", "systemd_logs"],
"privilege_escalation": ["windows_event_4672", "sudo_logs", "auditd", "edr_process_logs"],
"defense_evasion": ["edr_process_logs", "windows_event_4663", "sysmon_event_11", "antivirus_logs"],
"credential_access": ["windows_event_4625", "windows_event_4648", "lsass_access_events", "vault_audit_logs"],
"discovery": ["windows_event_4688", "auditd", "network_flow_logs", "dns_logs"],
"lateral_movement": ["windows_event_4624", "smb_logs", "winrm_logs", "network_flow_logs"],
"collection": ["dlp_alerts", "file_access_logs", "clipboard_monitoring", "screen_capture_logs"],
"command_and_control": ["dns_logs", "proxy_logs", "firewall_logs", "netflow_records"],
"exfiltration": ["dlp_alerts", "firewall_logs", "proxy_logs", "dns_logs"],
}
IOC_SWEEP_TARGETS = {
"ip": ["firewall_logs", "netflow_records", "proxy_logs", "threat_intel_platform"],
"domain": ["dns_logs", "proxy_logs", "email_gateway_logs", "threat_intel_platform"],
"hash": ["edr_hash_scanning", "antivirus_logs", "file_integrity_monitoring", "threat_intel_platform"],
"url": ["proxy_logs", "email_gateway_logs", "browser_history_logs"],
"email": ["email_gateway_logs", "dlp_alerts"],
"user_agent": ["proxy_logs", "web_application_logs"],
}
IOC_MAX_AGE_DAYS = 30 # IOCs older than this are flagged as stale
HUNT_KEYWORDS = {
"wmi": {"tactic": "lateral_movement", "mitre": "T1047", "data_source_key": "lateral_movement"},
"powershell": {"tactic": "execution", "mitre": "T1059.001", "data_source_key": "execution"},
"lolbin": {"tactic": "defense_evasion", "mitre": "T1218", "data_source_key": "defense_evasion"},
"lolbas": {"tactic": "defense_evasion", "mitre": "T1218", "data_source_key": "defense_evasion"},
"pass-the-hash": {"tactic": "lateral_movement", "mitre": "T1550.002", "data_source_key": "lateral_movement"},
"pth": {"tactic": "lateral_movement", "mitre": "T1550.002", "data_source_key": "lateral_movement"},
"credential dump": {"tactic": "credential_access", "mitre": "T1003", "data_source_key": "credential_access"},
"mimikatz": {"tactic": "credential_access", "mitre": "T1003.001", "data_source_key": "credential_access"},
"lateral": {"tactic": "lateral_movement", "mitre": "T1021", "data_source_key": "lateral_movement"},
"persistence": {"tactic": "persistence", "mitre": "T1053", "data_source_key": "persistence"},
"exfil": {"tactic": "exfiltration", "mitre": "T1041", "data_source_key": "exfiltration"},
"beacon": {"tactic": "command_and_control", "mitre": "T1071", "data_source_key": "command_and_control"},
"c2": {"tactic": "command_and_control", "mitre": "T1071", "data_source_key": "command_and_control"},
"ransomware": {"tactic": "impact", "mitre": "T1486", "data_source_key": "execution"},
"privilege": {"tactic": "privilege_escalation", "mitre": "T1068", "data_source_key": "privilege_escalation"},
"injection": {"tactic": "defense_evasion", "mitre": "T1055", "data_source_key": "defense_evasion"},
"apt": {"tactic": "initial_access", "mitre": "T1190", "data_source_key": "initial_access"},
"supply chain": {"tactic": "initial_access", "mitre": "T1195", "data_source_key": "initial_access"},
"phishing": {"tactic": "initial_access", "mitre": "T1566", "data_source_key": "initial_access"},
"scheduled task": {"tactic": "persistence", "mitre": "T1053", "data_source_key": "persistence"},
}
ANOMALY_TIME_HOURS_SUSPICIOUS = list(range(0, 6)) + list(range(22, 24))
# ---------------------------------------------------------------------------
# Hunt mode
# ---------------------------------------------------------------------------
def hunt_mode(args):
"""Score and prioritize a threat hunting hypothesis."""
hypothesis = args.hypothesis or ""
hypothesis_lower = hypothesis.lower()
# Extract T-code references via regex
matched_tcodes = list(set(re.findall(MITRE_PATTERN, hypothesis, re.IGNORECASE)))
# Keyword matching — multi-word keywords must be checked before single-word
matched_keywords = []
seen_keywords = set()
sorted_keywords = sorted(HUNT_KEYWORDS.keys(), key=lambda k: -len(k))
for kw in sorted_keywords:
if kw in hypothesis_lower and kw not in seen_keywords:
matched_keywords.append(kw)
seen_keywords.add(kw)
# Build tactic set from matched keywords and any T-codes that map to known tactics
tactics = set()
for kw in matched_keywords:
tactics.add(HUNT_KEYWORDS[kw]["tactic"])
# T-codes that happen to be in our keyword map (by mitre field)
for tcode in matched_tcodes:
for kw_data in HUNT_KEYWORDS.values():
if kw_data["mitre"].upper() == tcode.upper():
tactics.add(kw_data["tactic"])
break
# Collect data sources for matched tactics (deduped, ordered)
data_sources_set = []
seen_sources = set()
for tactic in tactics:
for src in HUNT_DATA_SOURCES.get(tactic, []):
if src not in seen_sources:
seen_sources.add(src)
data_sources_set.append(src)
# Scoring
actor_relevance = getattr(args, "actor_relevance", 1)
control_gap = getattr(args, "control_gap", 1)
data_availability = getattr(args, "data_availability", 2)
base_score = len(matched_keywords) * 2 + len(matched_tcodes) * 3
priority_score = base_score + actor_relevance * 3 + control_gap * 2 + data_availability
pursue_threshold = 5
pursue_recommendation = priority_score >= pursue_threshold
# Data quality check required if no data sources identified or low data_availability
data_quality_check_required = len(data_sources_set) == 0 or data_availability < 2
result = {
"mode": "hunt",
"hypothesis": hypothesis,
"matched_keywords": matched_keywords,
"matched_tcodes": matched_tcodes,
"tactics": sorted(tactics),
"data_sources_required": data_sources_set,
"priority_score": priority_score,
"pursue_recommendation": pursue_recommendation,
"data_quality_check_required": data_quality_check_required,
"score_breakdown": {
"base_score": base_score,
"actor_relevance_contribution": actor_relevance * 3,
"control_gap_contribution": control_gap * 2,
"data_availability_contribution": data_availability,
"pursue_threshold": pursue_threshold,
},
}
return result
# ---------------------------------------------------------------------------
# IOC mode
# ---------------------------------------------------------------------------
def ioc_mode(args):
"""Process IOC list and emit sweep targets with freshness check."""
ioc_file = getattr(args, "ioc_file", None)
ioc_date_str = getattr(args, "ioc_date", None)
if not ioc_file:
return {
"mode": "ioc",
"error": "--ioc-file is required for ioc mode",
}
try:
with open(ioc_file, "r", encoding="utf-8") as fh:
ioc_data = json.load(fh)
except FileNotFoundError:
return {"mode": "ioc", "error": f"IOC file not found: {ioc_file}"}
except json.JSONDecodeError as exc:
return {"mode": "ioc", "error": f"Invalid JSON in IOC file: {exc}"}
# Normalise: accept both plural and singular key names
type_key_map = {
"ip": ["ip", "ips"],
"domain": ["domain", "domains"],
"hash": ["hash", "hashes"],
"url": ["url", "urls"],
"email": ["email", "emails"],
"user_agent": ["user_agent", "user_agents"],
}
ioc_counts = {}
ioc_values = {} # type -> list of values
for ioc_type, candidate_keys in type_key_map.items():
for ck in candidate_keys:
if ck in ioc_data:
vals = ioc_data[ck]
if isinstance(vals, list) and vals:
ioc_counts[ioc_type] = len(vals)
ioc_values[ioc_type] = vals
break
# Freshness check
freshness_warning = False
ioc_age_days = None
if ioc_date_str:
try:
ioc_date = datetime.strptime(ioc_date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc)
now = datetime.now(tz=timezone.utc)
ioc_age_days = (now - ioc_date).days
if ioc_age_days > IOC_MAX_AGE_DAYS:
freshness_warning = True
except ValueError:
pass # invalid date format — skip freshness check
# Build sweep plan
sweep_plan = {}
for ioc_type, count in ioc_counts.items():
stale = freshness_warning # applies to entire IOC batch
sweep_plan[ioc_type] = {
"count": count,
"targets": IOC_SWEEP_TARGETS.get(ioc_type, []),
"stale": stale,
}
# Coverage score: ratio of represented IOC types to total possible
coverage_score = round(len(ioc_counts) / len(IOC_SWEEP_TARGETS), 4) if IOC_SWEEP_TARGETS else 0.0
# Recommended action
if freshness_warning:
recommended_action = (
"IOCs are stale (>{} days old). Re-validate against current threat intel feeds "
"before sweeping. Prioritise re-enrichment in threat intel platform.".format(IOC_MAX_AGE_DAYS)
)
elif not ioc_counts:
recommended_action = "No valid IOC types found in file. Verify JSON structure: expected keys ip, domain, hash, url, email."
elif coverage_score < 0.5:
recommended_action = (
"Partial IOC coverage ({:.0%}). Supplement with additional IOC types for broader detection fidelity. "
"Begin sweep in parallel.".format(coverage_score)
)
else:
recommended_action = (
"IOC set covers {:.0%} of sweep targets. Initiate concurrent sweep across all listed log sources. "
"Escalate any matches immediately.".format(coverage_score)
)
result = {
"mode": "ioc",
"ioc_counts": ioc_counts,
"sweep_plan": sweep_plan,
"coverage_score": coverage_score,
"freshness_warning": freshness_warning,
"ioc_age_days": ioc_age_days,
"recommended_action": recommended_action,
}
return result
# ---------------------------------------------------------------------------
# Anomaly mode
# ---------------------------------------------------------------------------
def anomaly_mode(args):
"""Z-score behavioral anomaly detection against a provided baseline."""
events_file = getattr(args, "events_file", None)
baseline_mean = getattr(args, "baseline_mean", None)
baseline_std = getattr(args, "baseline_std", None)
if not events_file:
return {"mode": "anomaly", "error": "--events-file is required for anomaly mode"}
if baseline_mean is None or baseline_std is None:
return {"mode": "anomaly", "error": "--baseline-mean and --baseline-std are required for anomaly mode"}
if baseline_std <= 0:
return {"mode": "anomaly", "error": "--baseline-std must be greater than 0"}
try:
with open(events_file, "r", encoding="utf-8") as fh:
events = json.load(fh)
except FileNotFoundError:
return {"mode": "anomaly", "error": f"Events file not found: {events_file}"}
except json.JSONDecodeError as exc:
return {"mode": "anomaly", "error": f"Invalid JSON in events file: {exc}"}
if not isinstance(events, list):
return {"mode": "anomaly", "error": "Events file must contain a JSON array of event objects"}
anomaly_events = []
soft_flag_count = 0
hard_flag_count = 0
time_anomaly_count = 0
entity_counts = {} # entity -> anomaly count
for idx, event in enumerate(events):
if not isinstance(event, dict):
continue
volume = event.get("volume")
timestamp_str = event.get("timestamp", "")
entity = event.get("entity", f"unknown_{idx}")
action = event.get("action", "")
# Z-score calculation
z_score = None
soft_flag = False
hard_flag = False
if volume is not None:
try:
volume = float(volume)
z_score = (volume - baseline_mean) / baseline_std
if z_score >= 3.0:
hard_flag = True
hard_flag_count += 1
entity_counts[entity] = entity_counts.get(entity, 0) + 1
elif z_score >= 2.0:
soft_flag = True
soft_flag_count += 1
entity_counts[entity] = entity_counts.get(entity, 0) + 1
except (TypeError, ValueError):
pass
# Time anomaly check
time_anomaly = False
event_hour = None
if timestamp_str:
for fmt in ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"):
try:
dt = datetime.strptime(timestamp_str, fmt)
event_hour = dt.hour
break
except ValueError:
continue
# Try with timezone offset via fromisoformat (Python 3.7+)
if event_hour is None:
try:
dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
event_hour = dt.hour
except ValueError:
pass
if event_hour is not None and event_hour in ANOMALY_TIME_HOURS_SUSPICIOUS:
time_anomaly = True
time_anomaly_count += 1
if soft_flag or hard_flag or time_anomaly:
anomaly_events.append({
"event_index": idx,
"entity": entity,
"action": action,
"timestamp": timestamp_str,
"volume": volume,
"z_score": round(z_score, 4) if z_score is not None else None,
"soft_flag": soft_flag,
"hard_flag": hard_flag,
"time_anomaly": time_anomaly,
"event_hour": event_hour,
})
total_events = len(events)
risk_score = round(hard_flag_count / total_events, 4) if total_events > 0 else 0.0
# Top anomalous entities
top_entities = sorted(entity_counts.items(), key=lambda x: -x[1])[:5]
# Recommended action
if hard_flag_count > 0:
recommended_action = (
"{} hard anomalies detected (z >= 3.0). Initiate threat hunt and review affected entities: {}. "
"Escalate to incident response if entity is high-value.".format(
hard_flag_count,
", ".join(e for e, _ in top_entities[:3]) if top_entities else "unknown"
)
)
elif soft_flag_count > 0:
recommended_action = (
"{} soft anomalies detected (z >= 2.0). Investigate {} for unusual activity patterns. "
"Cross-correlate with other log sources.".format(
soft_flag_count,
", ".join(e for e, _ in top_entities[:3]) if top_entities else "unknown"
)
)
elif time_anomaly_count > 0:
recommended_action = (
"No volume anomalies, but {} events occurred during suspicious hours (22:00-06:00). "
"Verify whether this activity is expected for the affected entities.".format(time_anomaly_count)
)
else:
recommended_action = "No anomalies detected. Baseline appears stable for the provided event set."
result = {
"mode": "anomaly",
"total_events": total_events,
"baseline_mean": baseline_mean,
"baseline_std": baseline_std,
"anomaly_events": anomaly_events,
"risk_score": risk_score,
"soft_flag_count": soft_flag_count,
"hard_flag_count": hard_flag_count,
"time_anomaly_count": time_anomaly_count,
"top_anomalous_entities": [{"entity": e, "anomaly_count": c} for e, c in top_entities],
"recommended_action": recommended_action,
}
return result
# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description=(
"Threat Signal Analyzer — Hunt hypothesis scoring, IOC sweep planning, "
"and behavioral anomaly detection."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"Examples:\n"
" python3 threat_signal_analyzer.py --mode hunt --hypothesis 'APT using WMI for lateral movement' --json\n"
" python3 threat_signal_analyzer.py --mode ioc --ioc-file iocs.json --ioc-date 2026-01-15 --json\n"
" python3 threat_signal_analyzer.py --mode anomaly --events-file events.json "
"--baseline-mean 45.0 --baseline-std 12.0 --json\n"
"\nExit codes:\n"
" 0 No high-priority findings\n"
" 1 Medium-priority signals detected\n"
" 2 High-priority findings confirmed"
),
)
parser.add_argument(
"--mode",
choices=["hunt", "ioc", "anomaly"],
required=True,
help="Analysis mode: hunt | ioc | anomaly",
)
# Hunt args
parser.add_argument("--hypothesis", type=str, help="[hunt] Free-text threat hypothesis")
parser.add_argument("--actor-relevance", type=int, choices=[0, 1, 2, 3], default=1,
dest="actor_relevance",
help="[hunt] Actor relevance score 0-3 (default: 1)")
parser.add_argument("--control-gap", type=int, choices=[0, 1, 2, 3], default=1,
dest="control_gap",
help="[hunt] Security control gap score 0-3 (default: 1)")
parser.add_argument("--data-availability", type=int, choices=[0, 1, 2, 3], default=2,
dest="data_availability",
help="[hunt] Data availability score 0-3 (default: 2)")
# IOC args
parser.add_argument("--ioc-file", type=str, dest="ioc_file",
help="[ioc] Path to JSON file with IOC lists (keys: ips, domains, hashes, urls, emails)")
parser.add_argument("--ioc-date", type=str, dest="ioc_date",
help="[ioc] Date IOCs were collected (YYYY-MM-DD) for freshness check")
# Anomaly args
parser.add_argument("--events-file", type=str, dest="events_file",
help="[anomaly] Path to JSON array of events with {timestamp, entity, action, volume}")
parser.add_argument("--baseline-mean", type=float, dest="baseline_mean",
help="[anomaly] Baseline mean for volume z-score calculation")
parser.add_argument("--baseline-std", type=float, dest="baseline_std",
help="[anomaly] Baseline standard deviation for z-score calculation")
# Output
parser.add_argument("--json", action="store_true", dest="output_json",
help="Output results as JSON")
args = parser.parse_args()
if args.mode == "hunt":
if not args.hypothesis:
parser.error("--hypothesis is required for hunt mode")
result = hunt_mode(args)
priority_score = result.get("priority_score", 0)
if args.output_json:
print(json.dumps(result, indent=2))
else:
print("\n=== THREAT HUNT ANALYSIS ===")
print(f"Hypothesis : {result['hypothesis']}")
print(f"Matched Keywords: {', '.join(result['matched_keywords']) or 'None'}")
print(f"Matched T-Codes : {', '.join(result['matched_tcodes']) or 'None'}")
print(f"Tactics : {', '.join(result['tactics']) or 'None'}")
print(f"Priority Score : {priority_score} (threshold: {result['score_breakdown']['pursue_threshold']})")
print(f"Pursue? : {'YES' if result['pursue_recommendation'] else 'NO'}")
print(f"Data Sources : {', '.join(result['data_sources_required']) or 'None identified'}")
print(f"Quality Check : {'Required' if result['data_quality_check_required'] else 'Not required'}")
# Exit codes: >= 8 = high, 5-7 = medium, < 5 = low
if priority_score >= 8:
sys.exit(2)
elif priority_score >= 5:
sys.exit(1)
sys.exit(0)
elif args.mode == "ioc":
if not args.ioc_file:
parser.error("--ioc-file is required for ioc mode")
result = ioc_mode(args)
if "error" in result:
if args.output_json:
print(json.dumps(result, indent=2))
else:
print(f"ERROR: {result['error']}", file=sys.stderr)
sys.exit(1)
if args.output_json:
print(json.dumps(result, indent=2))
else:
print("\n=== IOC SWEEP PLAN ===")
print(f"IOC Counts : {result['ioc_counts']}")
print(f"Coverage Score : {result['coverage_score']:.2%}")
print(f"Freshness Warn : {'YES — IOCs may be stale' if result['freshness_warning'] else 'No'}")
if result.get("ioc_age_days") is not None:
print(f"IOC Age (days) : {result['ioc_age_days']}")
print(f"\nAction: {result['recommended_action']}")
print("\nSweep Plan:")
for ioc_type, plan in result["sweep_plan"].items():
stale_tag = " [STALE]" if plan["stale"] else ""
print(f" {ioc_type:<12} {plan['count']} IOC(s){stale_tag} -> {', '.join(plan['targets'])}")
# Exit codes based on staleness and coverage
if result["freshness_warning"]:
sys.exit(1)
if result["coverage_score"] >= 0.5 and not result["freshness_warning"]:
sys.exit(0)
sys.exit(1)
elif args.mode == "anomaly":
if not args.events_file:
parser.error("--events-file is required for anomaly mode")
if args.baseline_mean is None or args.baseline_std is None:
parser.error("--baseline-mean and --baseline-std are required for anomaly mode")
result = anomaly_mode(args)
if "error" in result:
if args.output_json:
print(json.dumps(result, indent=2))
else:
print(f"ERROR: {result['error']}", file=sys.stderr)
sys.exit(1)
if args.output_json:
print(json.dumps(result, indent=2))
else:
print("\n=== ANOMALY DETECTION REPORT ===")
print(f"Total Events : {result['total_events']}")
print(f"Baseline Mean : {result['baseline_mean']}")
print(f"Baseline Std : {result['baseline_std']}")
print(f"Hard Flags : {result['hard_flag_count']} (z >= 3.0)")
print(f"Soft Flags : {result['soft_flag_count']} (z >= 2.0)")
print(f"Time Anomalies : {result['time_anomaly_count']}")
print(f"Risk Score : {result['risk_score']:.4f}")
if result["top_anomalous_entities"]:
print("\nTop Anomalous Entities:")
for entry in result["top_anomalous_entities"]:
print(f" {entry['entity']}: {entry['anomaly_count']} anomaly(s)")
print(f"\nAction: {result['recommended_action']}")
if result["anomaly_events"]:
print("\nFlagged Events (first 10):")
for ev in result["anomaly_events"][:10]:
flags = []
if ev["hard_flag"]:
flags.append("HARD")
if ev["soft_flag"]:
flags.append("SOFT")
if ev["time_anomaly"]:
flags.append("TIME")
print(
f" [{', '.join(flags)}] entity={ev['entity']} "
f"volume={ev['volume']} z={ev['z_score']} ts={ev['timestamp']}"
)
# Exit codes
hard_flags = result.get("hard_flag_count", 0)
soft_flags = result.get("soft_flag_count", 0)
time_anomalies = result.get("time_anomaly_count", 0)
if hard_flags > 0:
sys.exit(2)
elif soft_flags > 0 or time_anomalies > 0:
sys.exit(1)
sys.exit(0)
if __name__ == "__main__":
main()
Related skills
How it compares
Use threat-detection for proactive hunts and IOC sweeps; switch to incident-response skills after a breach is confirmed and containment is required.
FAQ
How is threat-detection different from incident response?
threat-detection focuses on proactively finding attacker activity through hypothesis-driven hunts, IOC analysis, and behavioral anomalies. Incident response handles containment and recovery after a confirmed breach.
What detection methods does threat-detection cover?
threat-detection covers hypothesis-driven threat hunting, IOC sweep generation, z-score anomaly detection on telemetry, and MITRE ATT&CK-mapped prioritization to surface signals automated controls missed.
Is Threat Detection safe to install?
skills.sh reports 2 of 3 security scanners passed. Review the Security Audits panel on this page before installing in production.