feat(report): tag findings in AI systems with validated MITRE ATLAS techniques

This commit is contained in:
Alex Schapiro 2026-10-28 20:05:13 +00:00
parent 2ea56755e6
commit 58e68a5db3
8 changed files with 118 additions and 0 deletions

View file

@ -132,6 +132,8 @@ _DETAIL_PRIORITY_KEYS = (
"cwe",
"attack_techniques",
"attack_source",
"atlas_techniques",
"atlas_source",
"cvss",
"filed_at",
"target",

68
strix/report/atlas.py Normal file
View file

@ -0,0 +1,68 @@
"""MITRE ATLAS technique ids for findings in AI systems (LLM apps, agents, RAG).
The catalog is a pinned snapshot of the ATLAS techniques, generated from
MITRE's ``atlas-data`` release.
"""
from __future__ import annotations
import json
import re
from functools import cache
from importlib.resources import files
from typing import Any
_TECHNIQUE_ID = re.compile(r"\bAML\.T\d{4}(?:\.\d{3})?\b", re.IGNORECASE)
@cache
def _catalog() -> dict[str, Any]:
raw = files("strix.report").joinpath("atlas_catalog.json").read_text(encoding="utf-8")
data: dict[str, Any] = json.loads(raw)
return data
def atlas_version() -> str:
return str(_catalog()["version"])
def technique(technique_id: str) -> dict[str, Any] | None:
entry = _catalog()["techniques"].get(technique_id)
return entry if isinstance(entry, dict) else None
def technique_url(technique_id: str) -> str:
return f"https://atlas.mitre.org/techniques/{technique_id}"
def normalize_atlas_techniques(raw: Any) -> tuple[list[str], list[str]]:
"""Parse ATLAS ids from a list or string and check them against the catalog.
Returns the ids in first-seen order and one error per id that is not a
known ATLAS technique.
"""
if raw is None:
return [], []
items = raw if isinstance(raw, list) else [raw]
techniques: list[str] = []
errors: list[str] = []
for item in items:
text = str(item or "").strip()
if not text:
continue
found = [match.upper() for match in _TECHNIQUE_ID.findall(text)]
if not found:
errors.append(f"invalid ATLAS technique: '{text}' (expected an id like 'AML.T0051')")
continue
for technique_id in found:
if technique_id in techniques:
continue
if technique(technique_id) is None:
errors.append(
f"unknown ATLAS technique: '{technique_id}' is not a "
f"MITRE ATLAS v{atlas_version()} technique"
)
continue
techniques.append(technique_id)
return techniques, errors

File diff suppressed because one or more lines are too long

View file

@ -544,6 +544,7 @@ def _result_properties(
"cve",
"cwe",
"attack_techniques",
"atlas_techniques",
"impact",
"technical_analysis",
"remediation_steps",

View file

@ -92,6 +92,7 @@ UPDATABLE_REPORT_FIELDS = frozenset(
"cve",
"cwe",
"attack_techniques",
"atlas_techniques",
"code_locations",
"http_exchange_ids",
"fix_verification",
@ -355,6 +356,7 @@ class ReportState:
cve: str | None = None,
cwe: str | None = None,
attack_techniques: list[str] | None = None,
atlas_techniques: list[str] | None = None,
code_locations: list[dict[str, Any]] | None = None,
http_exchange_ids: list[str] | None = None,
fix_verification: str | None = None,
@ -415,6 +417,8 @@ class ReportState:
report["cwe"] = cwe.strip()
if attack_techniques:
report["attack_techniques"] = list(attack_techniques)
if atlas_techniques:
report["atlas_techniques"] = list(atlas_techniques)
if code_locations:
report["code_locations"] = code_locations
if http_exchange_ids:

View file

@ -242,6 +242,7 @@ def render_vulnerability_md(report: dict[str, Any]) -> str: # noqa: PLR0912, PL
("CVE", report.get("cve")),
("CWE", report.get("cwe")),
("MITRE ATT&CK", ", ".join(report.get("attack_techniques") or [])),
("MITRE ATLAS", ", ".join(report.get("atlas_techniques") or [])),
]
cvss = report.get("cvss")
if cvss is not None:

View file

@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any
from agents import RunContextWrapper, function_tool
from strix.report.atlas import normalize_atlas_techniques
from strix.report.attack import normalize_attack_techniques
from strix.tools.nullish import clean_optional
from strix.tools.proxy.tools import existing_request_ids
@ -471,6 +472,12 @@ def _collect_update_changes( # noqa: PLR0912, PLR0915
if not attack_errors:
changes["attack_techniques"] = attack_techniques
if fields.get("atlas_techniques") is not None:
atlas_techniques, atlas_errors = normalize_atlas_techniques(fields.get("atlas_techniques"))
errors.extend(atlas_errors)
if not atlas_errors:
changes["atlas_techniques"] = atlas_techniques
raw_http_exchange_ids = fields.get("http_exchange_ids")
http_exchange_ids, http_exchange_errors = _normalize_http_exchange_ids(raw_http_exchange_ids)
errors.extend(http_exchange_errors)
@ -813,6 +820,7 @@ async def _do_create(
fix_verification: str | None = None,
fix_pr_body: str | None = None,
attack_techniques: list[str] | None = None,
atlas_techniques: list[str] | None = None,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
@ -857,6 +865,8 @@ async def _do_create(
errors.extend(identifier_errors)
normalized_attack_techniques, attack_errors = normalize_attack_techniques(attack_techniques)
errors.extend(attack_errors)
normalized_atlas_techniques, atlas_errors = normalize_atlas_techniques(atlas_techniques)
errors.extend(atlas_errors)
normalized_http_exchange_ids, http_exchange_errors = _normalize_http_exchange_ids(
http_exchange_ids
)
@ -920,6 +930,7 @@ async def _do_create(
"cve": cve,
"cwe": cwe,
"attack_techniques": normalized_attack_techniques or None,
"atlas_techniques": normalized_atlas_techniques or None,
"code_locations": parsed_locations,
"fix_verification": fix_verification,
"fix_pr_body": fix_pr_body,
@ -1014,6 +1025,7 @@ async def create_vulnerability_report(
cve: str | None = None,
cwe: str | None = None,
attack_techniques: list[str] | None = None,
atlas_techniques: list[str] | None = None,
code_locations: list[dict[str, Any]] | None = None,
http_exchange_ids: list[str] | None = None,
confidence_rationale: str | None = None,
@ -1316,6 +1328,12 @@ async def create_vulnerability_report(
after that (credentials it read, commands it ran, data it
took). Tag only techniques the evidence shows; leave out
techniques the finding merely makes possible. Omit when unsure.
atlas_techniques: MITRE ATLAS technique ids (e.g. ``["AML.T0051"]``)
when the target is an AI system (LLM app, agent, RAG pipeline)
and the PoC showed the attack, for example prompt injection
(AML.T0051), jailbreak (AML.T0054), system prompt extraction
(AML.T0056) or data leakage (AML.T0057). Same rule: only what the
evidence shows. Omit for other findings.
code_locations: White-box findings — list of location objects.
http_exchange_ids: Proxy request IDs that prove this finding.
Copy these IDs from ``list_requests`` or ``view_request``.
@ -1471,6 +1489,7 @@ async def create_vulnerability_report(
cve=cve,
cwe=cwe,
attack_techniques=attack_techniques,
atlas_techniques=atlas_techniques,
code_locations=code_locations,
http_exchange_ids=http_exchange_ids,
fix_verification=fix_verification,
@ -1507,6 +1526,7 @@ async def update_vulnerability_report(
cve: str | None = None,
cwe: str | None = None,
attack_techniques: list[str] | None = None,
atlas_techniques: list[str] | None = None,
code_locations: list[dict[str, Any]] | None = None,
http_exchange_ids: list[str] | None = None,
fix_verification: str | None = None,
@ -1582,6 +1602,7 @@ async def update_vulnerability_report(
cwe: Replacement CWE id.
attack_techniques: Replacement MITRE ATT&CK technique ids. Same
rule as on create: only what the evidence demonstrates.
atlas_techniques: Replacement MITRE ATLAS technique ids. Same rule.
code_locations: Replacement code locations.
http_exchange_ids: Replacement proxy request ids. Pass an empty
list to remove all linked exchanges.
@ -1629,6 +1650,7 @@ async def update_vulnerability_report(
"cve": cve,
"cwe": cwe,
"attack_techniques": attack_techniques,
"atlas_techniques": atlas_techniques,
"code_locations": code_locations,
"http_exchange_ids": http_exchange_ids,
"fix_verification": fix_verification,

19
tests/test_atlas.py Normal file
View file

@ -0,0 +1,19 @@
from strix.report.atlas import normalize_atlas_techniques, technique_url
def test_normalize_atlas_techniques_validates_ids() -> None:
ids, errors = normalize_atlas_techniques(["aml.t0051", "AML.T0051", "AML.T0054 jailbreak"])
assert ids == ["AML.T0051", "AML.T0054"]
assert errors == []
def test_normalize_atlas_techniques_rejects_unknown_and_malformed() -> None:
ids, errors = normalize_atlas_techniques(["AML.T9999", "T1190"])
assert ids == []
assert len(errors) == 2
assert "unknown ATLAS technique" in errors[0]
assert "invalid ATLAS technique" in errors[1]
def test_technique_url() -> None:
assert technique_url("AML.T0051.000") == "https://atlas.mitre.org/techniques/AML.T0051.000"