feat(report): tag findings with MITRE ATT&CK techniques

Reports accept an optional attack_techniques list, validated against a pinned
MITRE ATT&CK Enterprise v20.0 catalog of active techniques. The ids flow into
vulnerabilities.json, the markdown report, and the SARIF result properties.

strix/interface/cloud/render.py

@@ -130,6 +130,8 @@ _DETAIL_PRIORITY_KEYS = (
     "finding_type",
     "cve",
     "cwe",
+    "attack_techniques",
+    "attack_source",
     "cvss",
     "filed_at",
     "target",

strix/report/attack.py

@@ -0,0 +1,70 @@
+"""MITRE ATT&CK Enterprise technique ids for findings.
+
+The catalog is a pinned snapshot of the active (not revoked, not deprecated)
+Enterprise techniques, generated from MITRE's ``attack-stix-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"\bT\d{4}(?:\.\d{3})?\b", re.IGNORECASE)
+
+
+@cache
+def _catalog() -> dict[str, Any]:
+    raw = files("strix.report").joinpath("attack_catalog.json").read_text(encoding="utf-8")
+    data: dict[str, Any] = json.loads(raw)
+    return data
+
+
+def attack_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://attack.mitre.org/techniques/{technique_id.replace('.', '/')}/"
+
+
+def normalize_attack_techniques(raw: Any) -> tuple[list[str], list[str]]:
+    """Parse technique 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 an
+    active Enterprise 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 ATT&CK technique: '{text}' (expected an id like 'T1190' or 'T1059.007')"
+            )
+            continue
+        for technique_id in found:
+            if technique_id in techniques:
+                continue
+            if technique(technique_id) is None:
+                errors.append(
+                    f"unknown ATT&CK technique: '{technique_id}' is not an active "
+                    f"MITRE ATT&CK Enterprise v{attack_version()} technique"
+                )
+                continue
+            techniques.append(technique_id)
+    return techniques, errors

strix/report/sarif.py

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

strix/report/state.py

@@ -91,6 +91,7 @@ UPDATABLE_REPORT_FIELDS = frozenset(
         "method",
         "cve",
         "cwe",
+        "attack_techniques",
         "code_locations",
         "http_exchange_ids",
         "fix_verification",
@@ -353,6 +354,7 @@ class ReportState:
         method: str | None = None,
         cve: str | None = None,
         cwe: str | None = None,
+        attack_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,
@@ -411,6 +413,8 @@ class ReportState:
             report["cve"] = cve.strip()
         if cwe:
             report["cwe"] = cwe.strip()
+        if attack_techniques:
+            report["attack_techniques"] = list(attack_techniques)
         if code_locations:
             report["code_locations"] = code_locations
         if http_exchange_ids:

strix/report/writer.py

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

strix/tools/reporting/tool.py

@@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any
 
 from agents import RunContextWrapper, function_tool
 
+from strix.report.attack import normalize_attack_techniques
 from strix.tools.nullish import clean_optional
 from strix.tools.proxy.tools import existing_request_ids
 
@@ -462,6 +463,14 @@ def _collect_update_changes(  # noqa: PLR0912, PLR0915
     if cwe:
         changes["cwe"] = cwe
 
+    if fields.get("attack_techniques") is not None:
+        attack_techniques, attack_errors = normalize_attack_techniques(
+            fields.get("attack_techniques")
+        )
+        errors.extend(attack_errors)
+        if not attack_errors:
+            changes["attack_techniques"] = attack_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)
@@ -803,6 +812,7 @@ async def _do_create(
     confidence_rationale: str | None = None,
     fix_verification: str | None = None,
     fix_pr_body: str | None = None,
+    attack_techniques: list[str] | None = None,
     agent_id: str | None = None,
     agent_name: str | None = None,
 ) -> dict[str, Any]:
@@ -845,6 +855,8 @@ async def _do_create(
     errors.extend(_validate_fix_verification(parsed_locations, fix_verification))
     cve, cwe, identifier_errors = _validate_identifiers(cve, cwe)
     errors.extend(identifier_errors)
+    normalized_attack_techniques, attack_errors = normalize_attack_techniques(attack_techniques)
+    errors.extend(attack_errors)
     normalized_http_exchange_ids, http_exchange_errors = _normalize_http_exchange_ids(
         http_exchange_ids
     )
@@ -907,6 +919,7 @@ async def _do_create(
             "method": method,
             "cve": cve,
             "cwe": cwe,
+            "attack_techniques": normalized_attack_techniques or None,
             "code_locations": parsed_locations,
             "fix_verification": fix_verification,
             "fix_pr_body": fix_pr_body,
@@ -1000,6 +1013,7 @@ async def create_vulnerability_report(
     method: str | None = None,
     cve: str | None = None,
     cwe: str | None = None,
+    attack_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,
@@ -1296,6 +1310,12 @@ async def create_vulnerability_report(
         method: HTTP method when relevant.
         cve: ``CVE-YYYY-NNNNN`` if certain, else omit.
         cwe: ``CWE-NNN`` (most specific child) if certain, else omit.
+        attack_techniques: MITRE ATT&CK Enterprise technique ids (e.g.
+            ``["T1190", "T1552.001"]``) for what the PoC actually did:
+            the entry point it exploited and each step it demonstrated
+            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.
         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``.
@@ -1450,6 +1470,7 @@ async def create_vulnerability_report(
         method=method,
         cve=cve,
         cwe=cwe,
+        attack_techniques=attack_techniques,
         code_locations=code_locations,
         http_exchange_ids=http_exchange_ids,
         fix_verification=fix_verification,
@@ -1485,6 +1506,7 @@ async def update_vulnerability_report(
     method: str | None = None,
     cve: str | None = None,
     cwe: str | None = None,
+    attack_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,
@@ -1558,6 +1580,8 @@ async def update_vulnerability_report(
         method: Replacement HTTP method.
         cve: Replacement CVE id.
         cwe: Replacement CWE id.
+        attack_techniques: Replacement MITRE ATT&CK technique ids. Same
+            rule as on create: only what the evidence demonstrates.
         code_locations: Replacement code locations.
         http_exchange_ids: Replacement proxy request ids. Pass an empty
             list to remove all linked exchanges.
@@ -1604,6 +1628,7 @@ async def update_vulnerability_report(
         "method": method,
         "cve": cve,
         "cwe": cwe,
+        "attack_techniques": attack_techniques,
         "code_locations": code_locations,
         "http_exchange_ids": http_exchange_ids,
         "fix_verification": fix_verification,
@@ -1905,6 +1930,7 @@ async def _do_create_dependency(  # noqa: PLR0912, PLR0915
     fix_verification: str | None = None,
     contextual_cvss_breakdown: dict[str, str] | None = None,
     contextual_cvss_reasoning: str | None = None,
+    attack_techniques: list[str] | None = None,
     agent_id: str | None = None,
     agent_name: str | None = None,
 ) -> dict[str, Any]:
@@ -1935,6 +1961,9 @@ async def _do_create_dependency(  # noqa: PLR0912, PLR0915
         if cwe_err:
             errors.append(cwe_err)
 
+    normalized_attack_techniques, attack_errors = normalize_attack_techniques(attack_techniques)
+    errors.extend(attack_errors)
+
     fix_effort = (fix_effort or "").strip().lower()
     if fix_effort not in _VALID_FIX_EFFORT:
         errors.append(
@@ -2061,6 +2090,7 @@ async def _do_create_dependency(  # noqa: PLR0912, PLR0915
             cvss=cvss_score if advisory_cvss is not None else None,
             cve=parsed_cve,
             cwe=cwe,
+            attack_techniques=normalized_attack_techniques or None,
             code_locations=parsed_locations,
             fix_verification=fix_verification,
             finding_class="dependency_cve",
@@ -2121,6 +2151,7 @@ async def create_dependency_report(
     fix_verification: str | None = None,
     contextual_cvss_breakdown: dict[str, str] | None = None,
     contextual_cvss_reasoning: str | None = None,
+    attack_techniques: list[str] | None = None,
 ) -> str:
     """File a known-CVE dependency (SCA) finding — one report per CVE x package.
 
@@ -2185,6 +2216,10 @@ async def create_dependency_report(
         package_ecosystem: e.g. ``npm`` / ``pypi`` / ``maven`` / ``go``.
         fixed_version: First non-vulnerable version, if known.
         cwe: ``CWE-NNN`` (most specific) if certain, else omit.
+        attack_techniques: MITRE ATT&CK Enterprise technique ids the
+            reachability evidence demonstrates (e.g. ``["T1190"]`` for a
+            proven exploitable path from a public endpoint). Omit when the
+            evidence does not show one.
         advisory_cvss: **Required.** Published advisory base score
             (0.0-10.0) — read it off the advisory (trivy CVSS / NVD / GHSA).
             It is the published reference the finding is rated against and
@@ -2310,6 +2345,7 @@ async def create_dependency_report(
         fix_verification=fix_verification,
         contextual_cvss_breakdown=contextual_cvss_breakdown,
         contextual_cvss_reasoning=contextual_cvss_reasoning,
+        attack_techniques=attack_techniques,
         agent_id=agent_id,
         agent_name=agent_name,
     )
@@ -2338,6 +2374,7 @@ _REPORT_SUMMARY_FIELDS = (
     "finding_class",
     "cve",
     "cwe",
+    "attack_techniques",
     "target",
     "endpoint",
     "method",

tests/test_reporting_fields.py

@@ -2408,3 +2408,57 @@ async def test_delete_tool_carries_the_caller_identity(report_state: ReportState
     assert history[0]["agent_id"] == "aaaa1111"
     assert history[0]["agent_name"] == "Recon Agent"
     assert "update_vulnerability_report" in delete_vulnerability_report.description
+
+
+async def test_create_records_attack_techniques(report_state: ReportState) -> None:
+    result = await _do_create(
+        **_CONFIRMED_KWARGS,
+        attack_techniques=["t1190", "T1552.001 - Credentials In Files", "T1190"],
+    )
+
+    assert result["success"] is True
+    assert report_state.vulnerability_reports[0]["attack_techniques"] == ["T1190", "T1552.001"]
+
+
+async def test_create_rejects_unknown_attack_technique(report_state: ReportState) -> None:
+    result = await _do_create(**_CONFIRMED_KWARGS, attack_techniques=["T9999", "exploit"])
+
+    assert result["success"] is False
+    assert any("unknown ATT&CK technique: 'T9999'" in error for error in result["errors"])
+    assert any("invalid ATT&CK technique: 'exploit'" in error for error in result["errors"])
+    assert report_state.vulnerability_reports == []
+
+
+async def test_create_without_attack_techniques_omits_the_field(
+    report_state: ReportState,
+) -> None:
+    result = await _do_create(**_CONFIRMED_KWARGS)
+
+    assert result["success"] is True
+    assert "attack_techniques" not in report_state.vulnerability_reports[0]
+
+
+def test_update_replaces_attack_techniques(report_state: ReportState) -> None:
+    _seed_weak_report(report_state)
+
+    result = _do_update(
+        report_id="vuln-0009",
+        update_reason="The chained PoC read cloud credentials.",
+        fields={"attack_techniques": ["T1190", "T1552.005"]},
+    )
+
+    assert result["success"] is True
+    assert report_state.vulnerability_reports[0]["attack_techniques"] == ["T1190", "T1552.005"]
+
+
+def test_update_rejects_revoked_attack_technique(report_state: ReportState) -> None:
+    _seed_weak_report(report_state)
+
+    result = _do_update(
+        report_id="vuln-0009",
+        update_reason="Tag the technique.",
+        fields={"attack_techniques": ["T1086"]},
+    )
+
+    assert result["success"] is False
+    assert "attack_techniques" not in report_state.vulnerability_reports[0]

tests/test_sarif.py

@@ -371,3 +371,10 @@ def test_calibration_metadata_survives_into_result_properties(tmp_path: Path) ->
     assert strix["confidence_rationale"] == "Reproduced once out of three attempts."
     assert strix["severity_change_conditions"] == "Critical if the WAF rule is removed."
     assert strix["fix_verification"] == "Not retested."
+
+
+def test_write_sarif_carries_attack_techniques(tmp_path: Path) -> None:
+    write_sarif(tmp_path, [_finding(attack_techniques=["T1190", "T1059"])])
+    result = _read(tmp_path)["runs"][0]["results"][0]
+
+    assert result["properties"]["strix"]["attack_techniques"] == ["T1190", "T1059"]