81 lines
3.5 KiB
Python
81 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
TACTICS = {
|
|
"recon": ("TA0043", "Reconnaissance"),
|
|
"initial-access": ("TA0001", "Initial Access"),
|
|
"credential-access": ("TA0006", "Credential Access"),
|
|
"lateral-movement": ("TA0008", "Lateral Movement"),
|
|
"command-and-control": ("TA0011", "Command and Control"),
|
|
"exfiltration": ("TA0010", "Exfiltration"),
|
|
"network-spoofing": ("TA0006", "Credential Access"),
|
|
"dns-anomaly": ("TA0011", "Command and Control"),
|
|
"threat-intel": ("TA0011", "Command and Control"),
|
|
}
|
|
|
|
TECHNIQUES = {
|
|
"recon": ("T1595", "Active Scanning"),
|
|
"credential-access": ("T1110", "Brute Force"),
|
|
"lateral-movement": ("T1021", "Remote Services"),
|
|
"network-spoofing": ("T1557", "Adversary-in-the-Middle"),
|
|
"command-and-control": ("T1071", "Application Layer Protocol"),
|
|
"dns-anomaly": ("T1071.004", "DNS"),
|
|
"exfiltration": ("T1041", "Exfiltration Over C2 Channel"),
|
|
}
|
|
|
|
|
|
def classify(stage: str, summary: str = "", record: dict[str, Any] | None = None) -> list[dict[str, str]]:
|
|
"""Return conservative ATT&CK annotations for one network-observable signal."""
|
|
stage = str(stage or "").strip().lower()
|
|
text = f"{stage} {summary or ''}".lower()
|
|
record = record or {}
|
|
tactic = TACTICS.get(stage)
|
|
technique = TECHNIQUES.get(stage)
|
|
|
|
if stage == "initial-access":
|
|
if any(token in text for token in ("exploit", "cve-", "web application", "public-facing")):
|
|
technique = ("T1190", "Exploit Public-Facing Application")
|
|
elif any(token in text for token in ("phishing", "smtp", "malicious file")):
|
|
technique = ("T1566", "Phishing")
|
|
elif stage in {"command-and-control", "threat-intel"}:
|
|
if record.get("dns_query") or " dns" in text or "domain" in text:
|
|
technique = ("T1071.004", "DNS")
|
|
elif record.get("http_host") or "http" in text:
|
|
technique = ("T1071.001", "Web Protocols")
|
|
elif record.get("tls_sni") or record.get("quic_sni") or "tls" in text or "quic" in text:
|
|
technique = ("T1071", "Application Layer Protocol")
|
|
elif stage == "lateral-movement":
|
|
if "rdp" in text or int(record.get("dest_port") or 0) == 3389:
|
|
technique = ("T1021.001", "Remote Desktop Protocol")
|
|
elif "smb" in text or int(record.get("dest_port") or 0) in {139, 445}:
|
|
technique = ("T1021.002", "SMB/Windows Admin Shares")
|
|
elif "ssh" in text or int(record.get("dest_port") or 0) == 22:
|
|
technique = ("T1021.004", "SSH")
|
|
elif stage == "exfiltration":
|
|
if record.get("dns_query") or "dns" in text or "tunnel" in text:
|
|
technique = ("T1048", "Exfiltration Over Alternative Protocol")
|
|
|
|
if not tactic:
|
|
return []
|
|
item = {"tactic_id": tactic[0], "tactic": tactic[1]}
|
|
if technique:
|
|
item.update({"technique_id": technique[0], "technique": technique[1]})
|
|
return [item]
|
|
|
|
|
|
def merge(existing: list[dict[str, str]], additions: list[dict[str, str]], limit: int = 24) -> list[dict[str, str]]:
|
|
out: list[dict[str, str]] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
for item in list(existing or []) + list(additions or []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
key = (str(item.get("tactic_id") or ""), str(item.get("technique_id") or ""))
|
|
if key in seen or not key[0]:
|
|
continue
|
|
seen.add(key)
|
|
out.append({k: str(v) for k, v in item.items() if v not in (None, "")})
|
|
if len(out) >= limit:
|
|
break
|
|
return out
|