extract.py

# SPDX-FileCopyrightText: 2026 KustoKing / SecM8
# SPDX-License-Identifier: Apache-2.0

"""Per-asset MITRE coverage extractor.

Reads ``(envelope, payload)`` and returns a normalised
:class:`ExtractedCoverage` of ``(tactics, techniques, severity)``,
merging two sources:

1. ``envelope.metadata`` (rich authoring metadata, when present).
2. The asset-native payload location (where the platform itself
   stores MITRE attribution -- the source of truth).

The extractor exists because the corpus today is dominated by
collected detections that carry only ``metadata: { arm_name: ... }``
-- the rich authoring fields were never authored. Reading from the
payload makes ``contentops coverage`` reflect what the platform
actually has, without requiring a hand-backfill of every YAML.

Per-asset payload locations:

* ``defender_custom_detection``:
    techniques = ``payload.detectionAction.alertTemplate.mitreTechniques``
    severity   = ``payload.detectionAction.alertTemplate.severity``
    tactics    = derived from techniques via the curated MITRE map;
                 fallback to ``alertTemplate.category`` if it matches
                 a canonical tactic name.
* ``sentinel_analytic``:
    tactics    = ``payload.tactics`` (PascalCase, matches the canonical Literal)
    techniques = ``payload.techniques``
    severity   = ``payload.severity`` (TitleCase -> lowercased)
* ``sentinel_hunting``:
    tactics    = ``payload.tactics``
    techniques = ``payload.techniques``
    severity   = ``"informational"`` (hunting queries don't carry severity)

When metadata AND payload both have data, the extractor unions them
(authored content is not silently dropped). When neither has data,
the triple is empty and the detection contributes only to the
``total_detections`` count, not to any per-tactic bucket.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
from typing import Any

from contentops.core.asset import Asset
from contentops.core.envelope import EnvelopeV2


_CANONICAL_TACTICS: frozenset[str] = frozenset({
    "Reconnaissance",
    "ResourceDevelopment",
    "InitialAccess",
    "Execution",
    "Persistence",
    "PrivilegeEscalation",
    "DefenseEvasion",
    "CredentialAccess",
    "Discovery",
    "LateralMovement",
    "Collection",
    "CommandAndControl",
    "Exfiltration",
    "Impact",
    # ARM Microsoft.SecurityInsights/alertRules tactic enum members
    # outside the canonical 14-tactic ATT&CK Enterprise list. Without
    # these, Sentinel rules carrying ``PreAttack`` (common on legacy
    # built-in templates) or the two ICS/OT tactics would be silently
    # dropped by the per-asset readers.
    "PreAttack",
    "ImpairProcessControl",
    "InhibitResponseFunction",
})

_CANONICAL_SEVERITIES: frozenset[str] = frozenset(
    {"informational", "low", "medium", "high"}
)

_DEFAULT_SEVERITY = "informational"


@dataclass(frozen=True)
class ExtractedCoverage:
    """Normalised MITRE coverage triple for one envelope.

    ``tactics`` and ``techniques`` always contain only canonical /
    well-formed values; case is normalised; duplicates removed.
    ``severity`` is always one of the four canonical lowercase
    values (defaults to ``"informational"``).

    ``techniques_without_tactic`` lists technique IDs that the
    extractor saw on the envelope but could not map to any tactic --
    typically because the technique is outside the bundled curated
    list at ``contentops/coverage/data/mitre_attack_techniques.json``.
    Surfaced so the operator can see the gap and supply a
    ``--techniques-file`` if needed.
    """

    tactics: tuple[str, ...] = field(default_factory=tuple)
    techniques: tuple[str, ...] = field(default_factory=tuple)
    severity: str = _DEFAULT_SEVERITY
    techniques_without_tactic: tuple[str, ...] = field(default_factory=tuple)


# ---------------------------------------------------------------------------
# Curated technique -> tactic lookup
# ---------------------------------------------------------------------------


@lru_cache(maxsize=1)
def _technique_to_tactics() -> dict[str, tuple[str, ...]]:
    """Return ``{technique_id: (tactic, ...)}`` from the bundled curated list.

    Loaded once per process. The bundled JSON is a curated subset
    (~70 high-value techniques). Techniques not in the list will have
    no tactic mapping and will be reported via
    ``ExtractedCoverage.techniques_without_tactic``.
    """
    data_path = Path(__file__).parent / "data" / "mitre_attack_techniques.json"
    raw = json.loads(data_path.read_text(encoding="utf-8"))
    out: dict[str, tuple[str, ...]] = {}
    for entry in raw.get("techniques", []):
        tid = entry.get("id")
        tactics = entry.get("tactics", [])
        if isinstance(tid, str) and isinstance(tactics, list):
            out[tid] = tuple(t for t in tactics if t in _CANONICAL_TACTICS)
    return out


# ---------------------------------------------------------------------------
# Per-asset payload readers
# ---------------------------------------------------------------------------

# Per-asset readers all return the same 4-tuple shape:
#   (tactics, techniques, severity, techniques_without_tactic)
# tactics/techniques/orphans are lists; severity is the lowercase
# canonical value or ``None`` (caller falls through to default).

_ReaderResult = tuple[list[str], list[str], "str | None", list[str]]


def _str_list(value: Any) -> list[str]:
    """Coerce a payload field to a clean ``list[str]``; tolerant of None / non-list."""
    if not isinstance(value, list):
        return []
    return [v for v in value if isinstance(v, str) and v]


def _normalise_severity(value: Any) -> "str | None":
    """Return one of the canonical severities, or ``None`` if unrecognised."""
    if not isinstance(value, str) or not value:
        return None
    lowered = value.strip().lower()
    if lowered in _CANONICAL_SEVERITIES:
        return lowered
    return None


def _defender_payload(payload: dict[str, Any]) -> _ReaderResult:
    """Read MITRE data from a defender_custom_detection payload."""
    detection_action = payload.get("detectionAction")
    alert: dict[str, Any] = {}
    if isinstance(detection_action, dict):
        candidate = detection_action.get("alertTemplate")
        if isinstance(candidate, dict):
            alert = candidate

    techniques = _str_list(alert.get("mitreTechniques"))
    severity = _normalise_severity(alert.get("severity"))

    # Defender doesn't store tactics directly. Derive from techniques
    # via the curated map; fall back to ``category`` only when it
    # matches a canonical tactic name (common case: "Discovery").
    lookup = _technique_to_tactics()
    tactics: list[str] = []
    orphans: list[str] = []
    for tid in techniques:
        mapped = lookup.get(tid)
        if mapped:
            tactics.extend(mapped)
        else:
            orphans.append(tid)

    if not tactics:
        category = alert.get("category")
        if isinstance(category, str) and category in _CANONICAL_TACTICS:
            tactics = [category]
            # The category-fallback "covers" the orphans for tactic
            # purposes -- they did contribute to a bucket via category
            # -- so don't report them as orphans in this case.
            orphans = []

    return tactics, techniques, severity, orphans


def _sentinel_analytic_payload(payload: dict[str, Any]) -> _ReaderResult:
    tactics_raw = _str_list(payload.get("tactics"))
    tactics = [t for t in tactics_raw if t in _CANONICAL_TACTICS]
    techniques = _str_list(payload.get("techniques"))
    severity = _normalise_severity(payload.get("severity"))
    return tactics, techniques, severity, []


def _sentinel_hunting_payload(payload: dict[str, Any]) -> _ReaderResult:
    """Hunting queries don't carry a severity; return ``None`` so the
    caller falls through to the default.
    """
    tactics_raw = _str_list(payload.get("tactics"))
    tactics = [t for t in tactics_raw if t in _CANONICAL_TACTICS]
    techniques = _str_list(payload.get("techniques"))
    return tactics, techniques, None, []


_PAYLOAD_READERS = {
    Asset.DEFENDER_CUSTOM_DETECTION: _defender_payload,
    Asset.SENTINEL_ANALYTIC: _sentinel_analytic_payload,
    Asset.SENTINEL_HUNTING: _sentinel_hunting_payload,
}


# ---------------------------------------------------------------------------
# Top-level extractor
# ---------------------------------------------------------------------------


def extract_mitre(envelope: EnvelopeV2, payload: dict[str, Any]) -> ExtractedCoverage:
    """Return the normalised coverage triple for one envelope.

    Combines two sources:

    1. ``envelope.metadata`` (priority for severity; union for tactics
       + techniques).
    2. Asset-native payload reader (for the typical case where rich
       metadata was never authored).

    When both sources have data, tactics + techniques are unioned and
    sorted; severity prefers the metadata value.

    Returns an empty (but valid) :class:`ExtractedCoverage` if the
    asset is not a detection kind or has no MITRE data anywhere.
    """
    reader = _PAYLOAD_READERS.get(envelope.asset)
    if reader is None:
        return ExtractedCoverage()

    payload_tactics, payload_techniques, payload_sev, orphans = reader(payload)

    meta_tactics: list[str] = []
    meta_techniques: list[str] = []
    meta_sev: "str | None" = None
    if envelope.metadata is not None:
        meta_tactics = list(envelope.metadata.tactics)
        meta_techniques = list(envelope.metadata.techniques)
        meta_sev = envelope.metadata.severity

    merged_tactics = tuple(sorted(set(meta_tactics + payload_tactics)))
    merged_techniques = tuple(sorted(set(meta_techniques + payload_techniques)))
    severity = meta_sev or payload_sev or _DEFAULT_SEVERITY

    return ExtractedCoverage(
        tactics=merged_tactics,
        techniques=merged_techniques,
        severity=severity,
        techniques_without_tactic=tuple(orphans),
    )


__all__ = [
    "ExtractedCoverage",
    "extract_mitre",
]