WIP - Initial MITRE v19 Support

CLI.md

@@ -642,6 +642,27 @@ rules. This is based on the hash of the rule in the following format:
 
 As a result, all cases where rules are shown or converted to JSON are not just simple conversions from TOML.
 
+
+## Multi-version threat mappings (MITRE ATT&CK v18 / v19)
+
+Rules can carry more than one ATT&CK mapping at once. The `threat` field holds the baseline mapping
+(MITRE ATT&CK v18) that ships to Kibana, while an optional `threat_mappings` field holds additional
+version-tagged mappings (e.g. v19). At build time exactly one mapping is emitted as the API `threat`
+(default: the baseline), selected via the `threat_mapping_framework` / `threat_mapping_version` config
+keys or the `DR_THREAT_MAPPING_FRAMEWORK` / `DR_THREAT_MAPPING_VERSION` environment variables;
+`threat_mappings` is always stripped from the shipped artifact.
+
+Generate a target-version mapping from a rule's existing mapping with
+`dev attack convert-threat-mappings` (accuracy-first: anything not present in the mapping config is
+dropped, never guessed), and scaffold a mapping config with `dev attack scaffold-version-map`. The
+feature is DaC-aware. See [docs-dev/multi-version-threat-mappings.md](docs-dev/multi-version-threat-mappings.md)
+for the full guide.
+
+```bash
+# preview v18 -> v19 conversion without writing
+python -m detection_rules dev attack convert-threat-mappings -t 19 --dry-run
+```
+
 ## Debugging
 
 Most of the CLI errors will print a concise, user friendly error. To enable debug mode and see full error stacktraces,

detection_rules/attack.py

@@ -6,20 +6,24 @@
 """Mitre attack info."""
 
 import json
+import os
 import re
 import time
 from collections import OrderedDict
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
 import requests
+import yaml
 from semver import Version
 
 from .utils import cached, clear_caches, get_etc_glob_path, get_etc_path, gzip_compress, read_gzip
 
 PLATFORMS = ["Windows", "macOS", "Linux"]
 CROSSWALK_FILE = get_etc_path(["attack-crosswalk.json"])
 TECHNIQUES_REDIRECT_FILE = get_etc_path(["attack-technique-redirects.json"])
+ATTACK_VERSION_MAPS_DIRNAME = "attack-version-maps"
 
 tactics_map: dict[str, Any] = {}
 
@@ -29,14 +33,30 @@ def load_techniques_redirect() -> dict[str, Any]:
     return json.loads(TECHNIQUES_REDIRECT_FILE.read_text())["mapping"]
 
 
+def _attack_file_version(path: Path) -> Version:
+    """Extract the semver from an ATT&CK gz filename for sorting."""
+    ver = path.name.split("-v", 1)[1][: -len(".json.gz")]
+    return Version.parse(ver, optional_minor_and_patch=True)
+
+
 def get_attack_file_path() -> Path:
+    """Return the baseline ATT&CK data file (lowest available version)."""
     pattern = "attack-v*.json.gz"
-    attack_file = get_etc_glob_path([pattern])
-    if len(attack_file) < 1:
+    attack_files = get_etc_glob_path([pattern])
+    if not attack_files:
         raise FileNotFoundError(f"Missing required {pattern} file")
-    if len(attack_file) != 1:
-        raise FileExistsError(f"Multiple files found with {pattern} pattern. Only one is allowed")
-    return Path(attack_file[0])
+    return min(attack_files, key=_attack_file_version)
+
+
+def get_attack_file_path_for_version(version: str) -> Path:
+    """Return the ATT&CK data file whose major version matches ``version``."""
+    major = version.split(".", maxsplit=1)[0]
+    attack_files = get_etc_glob_path(["attack-v*.json.gz"])
+    for path in sorted(attack_files, key=_attack_file_version, reverse=True):
+        if path.name.split("-v", 1)[1][: -len(".json.gz")].split(".")[0] == major:
+            return path
+    available = [p.name.split("-v", 1)[1][: -len(".json.gz")] for p in attack_files]
+    raise FileNotFoundError(f"No ATT&CK data file found for version {version!r}. Available: {available}")
 
 
 _, _attack_path_base = str(get_attack_file_path()).split("-v")
@@ -98,6 +118,76 @@ def load_attack_gz() -> dict[str, Any]:
 technique_id_list = [t for t in technique_lookup if "." not in t]
 sub_technique_id_list = [t for t in technique_lookup if "." in t]
 
+# Index of all ATT&CK gz files present in etc/, keyed by full version string (e.g. "18.1.0", "19.1").
+AVAILABLE_ATTACK_VERSIONS: dict[str, Path] = {
+    p.name.split("-v", 1)[1][: -len(".json.gz")]: p for p in get_etc_glob_path(["attack-v*.json.gz"])
+}
+
+
+@dataclass
+class AttackLookups:
+    """Pre-built ATT&CK lookup structures for a specific version."""
+
+    version: str
+    tactics_map: dict[str, str]
+    technique_lookup: OrderedDict  # type: ignore[type-arg]
+    revoked: dict[str, Any]
+    deprecated: dict[str, Any]
+    matrix: dict[str, list[str]]
+
+
+def _build_lookups(version: str, raw: dict[str, Any]) -> AttackLookups:
+    """Build ATT&CK lookup structures from raw STIX data."""
+    _tactics_map: dict[str, str] = {}
+    _technique_lookup: dict[str, Any] = {}
+    _revoked: dict[str, Any] = {}
+    _deprecated: dict[str, Any] = {}
+
+    for item in raw["objects"]:
+        if item["type"] == "x-mitre-tactic":
+            _tactics_map[item["name"]] = item["external_references"][0]["external_id"]
+        if item["type"] == "attack-pattern" and item["external_references"][0]["source_name"] == "mitre-attack":
+            tid = item["external_references"][0]["external_id"]
+            _technique_lookup[tid] = item
+            if item.get("revoked"):
+                _revoked[tid] = item
+            if item.get("x_mitre_deprecated"):
+                _deprecated[tid] = item
+
+    _tactics_list = list(_tactics_map)
+    _matrix: dict[str, list[str]] = {t: [] for t in _tactics_list}
+    for tid, technique in sorted(_technique_lookup.items(), key=lambda kv: kv[1]["name"].lower()):
+        kill_chain = technique.get("kill_chain_phases")
+        if kill_chain:
+            for phase in kill_chain:
+                if phase["kill_chain_name"] != "mitre-attack":
+                    continue
+                tactic_name = next(
+                    (t for t in _tactics_list if t.lower() == phase["phase_name"].replace("-", " ")),
+                    None,
+                )
+                if tactic_name:
+                    _matrix[tactic_name].append(tid)
+    for val in _matrix.values():
+        val.sort(key=lambda t: _technique_lookup[t]["name"].lower())
+
+    return AttackLookups(
+        version=version,
+        tactics_map=_tactics_map,
+        technique_lookup=OrderedDict(sorted(_technique_lookup.items())),
+        revoked=dict(sorted(_revoked.items())),
+        deprecated=dict(sorted(_deprecated.items())),
+        matrix=_matrix,
+    )
+
+
+@cached
+def build_attack_lookups_for_version(version: str) -> AttackLookups:
+    """Load and cache ATT&CK lookup structures for a specific version."""
+    path = get_attack_file_path_for_version(version)
+    raw = json.loads(read_gzip(path))
+    return _build_lookups(version, raw)
+
 
 def refresh_attack_data(save: bool = True) -> tuple[dict[str, Any] | None, bytes | None]:
     """Refresh ATT&CK data from Mitre."""
@@ -248,3 +338,202 @@ def refresh_redirected_techniques_map(threads: int = 50) -> None:
 def load_crosswalk_map() -> dict[str, Any]:
     """Retrieve the replacement mapping."""
     return json.loads(CROSSWALK_FILE.read_text())["mapping"]
+
+
+# Multi-version threat mapping support. Generation of a target-version mapping from a source mapping
+# is driven by a curated, self-contained config (see etc/attack-version-maps/); ids absent from the
+# config (or mapped to null) are dropped, so generation is accuracy-first and never invents a mapping.
+
+# kinds of threat entry that can be remapped, with the corresponding config table attribute
+_MAP_KIND_ATTRS: dict[str, str] = {
+    "tactic": "tactics",
+    "technique": "techniques",
+    "subtechnique": "subtechniques",
+}
+_REQUIRED_MAP_KEYS = ("framework", "source_version", "target_version")
+
+
+@dataclass
+class AttackVersionMap:
+    """A curated source->target ATT&CK (or other framework) mapping config."""
+
+    framework: str
+    source_version: str
+    target_version: str
+    path: Path
+    tactics: dict[str, dict[str, str] | None] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+    techniques: dict[str, dict[str, str] | None] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+    subtechniques: dict[str, dict[str, str] | None] = field(default_factory=dict)  # type: ignore[reportUnknownVariableType]
+
+    @property
+    def key(self) -> tuple[str, str, str]:
+        return (self.framework, str(self.source_version), str(self.target_version))
+
+    def _table(self, kind: str) -> dict[str, dict[str, str] | None]:
+        attr = _MAP_KIND_ATTRS.get(kind)
+        if attr is None:
+            raise ValueError(f"Unknown threat entry kind: {kind}")
+        return getattr(self, attr)
+
+    def is_mapped(self, kind: str, source_id: str) -> bool:
+        """Whether the source id has an explicit, non-null destination in the config."""
+        return self._table(kind).get(source_id) is not None
+
+    def lookup(self, kind: str, source_id: str) -> dict[str, str] | None:
+        """Return the destination entry for a source id, or ``None`` if dropped/absent."""
+        return self._table(kind).get(source_id)
+
+
+def parse_attack_version_map(path: Path) -> AttackVersionMap:
+    """Parse a single attack version mapping config file."""
+    raw: dict[str, Any] = yaml.safe_load(path.read_text()) or {}
+    missing = [k for k in _REQUIRED_MAP_KEYS if not raw.get(k)]
+    if missing:
+        raise ValueError(f"Attack version map {path} is missing required keys: {missing}")
+
+    def _validate_table(table: dict[str, Any], kind: str) -> dict[str, dict[str, str] | None]:
+        for source_id, dest in table.items():
+            if dest is None:
+                continue
+            if not isinstance(dest, dict) or not all(k in dest for k in ("id", "name", "reference")):
+                raise ValueError(
+                    f"Attack version map {path}: {kind} entry '{source_id}' must map to null or an "
+                    f"object with 'id', 'name', and 'reference' (got: {dest!r})"
+                )
+        return table
+
+    return AttackVersionMap(
+        framework=str(raw["framework"]),
+        source_version=str(raw["source_version"]),
+        target_version=str(raw["target_version"]),
+        path=path,
+        tactics=_validate_table(raw.get("tactics") or {}, "tactics"),
+        techniques=_validate_table(raw.get("techniques") or {}, "techniques"),
+        subtechniques=_validate_table(raw.get("subtechniques") or {}, "subtechniques"),
+    )
+
+
+def load_attack_version_maps(paths: list[Path]) -> dict[tuple[str, str, str], AttackVersionMap]:
+    """Load mapping configs from explicit files and/or directories, indexed by (framework, src, tgt)."""
+    files: list[Path] = []
+    for p in paths:
+        if p.is_dir():
+            files.extend(sorted(set(p.glob("*.yaml")) | set(p.glob("*.yml"))))
+        elif p.exists():
+            files.append(p)
+        else:
+            raise FileNotFoundError(f"Attack version map path not found: {p}")
+
+    maps: dict[tuple[str, str, str], AttackVersionMap] = {}
+    for f in files:
+        parsed = parse_attack_version_map(f)
+        if parsed.key in maps:
+            raise ValueError(f"Duplicate attack version map for {parsed.key}: {f} and {maps[parsed.key].path}")
+        maps[parsed.key] = parsed
+    return maps
+
+
+def get_attack_version_map(
+    framework: str,
+    source_version: str,
+    target_version: str,
+    paths: list[Path] | None = None,
+) -> AttackVersionMap:
+    """Return the mapping config matching the (framework, source, target) triple."""
+    if paths is None:
+        from .config import parse_rules_config
+
+        cfg_dir = parse_rules_config().attack_version_maps_dir
+        if not cfg_dir:
+            raise FileNotFoundError(
+                "No attack version maps directory configured (set `attack_version_maps_dir` in "
+                "_config.yaml) and no explicit config path was provided."
+            )
+        paths = [cfg_dir]
+
+    maps = load_attack_version_maps(paths)
+    key = (framework, str(source_version), str(target_version))
+    if key not in maps:
+        available = sorted(maps)
+        raise ValueError(
+            f"No attack version map found for framework={framework!r} {source_version}->{target_version}. "
+            f"Available: {available}"
+        )
+    return maps[key]
+
+
+def build_identity_version_map(framework: str, source_version: str, target_version: str) -> dict[str, Any]:
+    """Build a cross-version identity skeleton using source IDs and target-version names."""
+    # Source data: determines which IDs appear as source keys.
+    try:
+        src = build_attack_lookups_for_version(source_version)
+        _src_tactics_map = src.tactics_map
+        _src_technique_lookup = src.technique_lookup
+        _src_revoked = src.revoked
+        _src_deprecated = src.deprecated
+    except FileNotFoundError:
+        _src_tactics_map = tactics_map
+        _src_technique_lookup = technique_lookup
+        _src_revoked = revoked
+        _src_deprecated = deprecated
+
+    # Target data: resolves destination names/references where the same ID exists.
+    try:
+        tgt = build_attack_lookups_for_version(target_version)
+        _tgt_tactic_id_to_name: dict[str, str] = {v: k for k, v in tgt.tactics_map.items()}
+        _tgt_technique_lookup: dict[str, Any] = dict(tgt.technique_lookup)
+    except FileNotFoundError:
+        _tgt_tactic_id_to_name = {}
+        _tgt_technique_lookup = {}
+
+    url_base = "https://attack.mitre.org/{type}/{id}/"
+
+    # Tactics: source IDs as keys, target name if the ID still exists in the target version.
+    out_tactics: dict[str, dict[str, str]] = {
+        tactic_id: {
+            "id": tactic_id,
+            "name": _tgt_tactic_id_to_name.get(tactic_id, name),
+            "reference": url_base.format(type="tactics", id=tactic_id),
+        }
+        for name, tactic_id in _src_tactics_map.items()
+    }
+
+    # Techniques and subtechniques: source IDs as keys, target name if available.
+    out_techniques: dict[str, dict[str, str]] = {}
+    out_subtechniques: dict[str, dict[str, str]] = {}
+    for technique_id, item in _src_technique_lookup.items():
+        if technique_id in _src_revoked or technique_id in _src_deprecated:
+            continue
+        tgt_item = _tgt_technique_lookup.get(technique_id, item)
+        entry: dict[str, str] = {
+            "id": technique_id,
+            "name": tgt_item["name"],
+            "reference": url_base.format(type="techniques", id=technique_id.replace(".", "/")),
+        }
+        if "." in technique_id:
+            out_subtechniques[technique_id] = entry
+        else:
+            out_techniques[technique_id] = entry
+
+    return {
+        "framework": framework,
+        "source_version": source_version,
+        "target_version": target_version,
+        "tactics": dict(sorted(out_tactics.items())),
+        "techniques": dict(sorted(out_techniques.items())),
+        "subtechniques": dict(sorted(out_subtechniques.items())),
+    }
+
+
+def resolve_output_threat_version() -> tuple[str, str]:
+    """Resolve which (framework, version) threat mapping should be emitted as the API `threat`."""
+    from .config import (
+        THREAT_MAPPING_FRAMEWORK_ENV,
+        THREAT_MAPPING_VERSION_ENV,
+        parse_rules_config,
+    )
+
+    cfg = parse_rules_config()
+    framework = os.getenv(THREAT_MAPPING_FRAMEWORK_ENV, cfg.threat_mapping_framework)
+    version = str(os.getenv(THREAT_MAPPING_VERSION_ENV, cfg.threat_mapping_version))
+    return framework, version

detection_rules/cli_utils.py

@@ -215,15 +215,18 @@ def get_collection(*args: Any, **kwargs: Any) -> Any:
 
         # Warn that if the path does not match the expected path, it will be saved to the expected path
         for rule in rules:
-            threat = rule.contents.data.get("threat")
-            first_tactic = threat[0].tactic.name if threat else ""
             # Check if flag or config is set to not include tactic in the filename
             no_tactic_filename = no_tactic_filename or RULES_CONFIG.no_tactic_filename
-            tactic_name = None if no_tactic_filename else first_tactic
-            rule_name = rulename_to_filename(rule.contents.data.name, tactic_name=tactic_name)
+            # Accept a filename built from the baseline tactic or any versioned (e.g. v19) threat_mappings
+            # tactic, since a rule may still use its older tactic name in the filename until it's renamed.
+            tactic_names = [] if no_tactic_filename else rule.contents.data.get_primary_tactic_names()
+            candidate_names = [rulename_to_filename(rule.contents.data.name, tactic_name=t) for t in tactic_names]
+            if not candidate_names:
+                candidate_names = [rulename_to_filename(rule.contents.data.name, tactic_name=None)]
+            rule_name = candidate_names[0]
             if not rule.path:
                 click.secho(f"WARNING: Rule path for rule not found: {rule_name}", fg="yellow")
-            elif rule.path.name != rule_name:
+            elif rule.path.name not in candidate_names:
                 click.secho(
                     f"WARNING: Rule path does not match required path: {rule.path.name} != {rule_name}", fg="yellow"
                 )