Security Planning

scripts/tm7_threat_contract.py

#!/usr/bin/env python3
# Copyright (c) 2026 Microsoft Corporation. All rights reserved.
# SPDX-License-Identifier: MIT
"""Validate and serialize the TM7 threat-instance DataContract."""

from __future__ import annotations

import hashlib
import re
import threading
import uuid
from collections import Counter
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import Any
from xml.etree import ElementTree as ET

from defusedxml import ElementTree as DefusedET

ABSTRACT_NS = "http://schemas.datacontract.org/2004/07/ThreatModeling.Model.Abstracts"
ARRAYS_NS = "http://schemas.microsoft.com/2003/10/Serialization/Arrays"
KNOWLEDGE_NS = "http://schemas.datacontract.org/2004/07/ThreatModeling.KnowledgeBase"
MODEL_NS = "http://schemas.datacontract.org/2004/07/ThreatModeling.Model"
SERIALIZATION_NS = "http://schemas.microsoft.com/2003/10/Serialization/"
XSD_NS = "http://www.w3.org/2001/XMLSchema"
XSI_NS = "http://www.w3.org/2001/XMLSchema-instance"

# TMT reads prefixes positionally in places, so a written model must carry the
# exact prefix bindings below rather than the ns0/ns1 names ElementTree invents.
TM7_NAMESPACE_PREFIXES: tuple[tuple[str, str], ...] = (
    ("", MODEL_NS),
    ("a", ARRAYS_NS),
    ("b", KNOWLEDGE_NS),
    ("d2p1", ABSTRACT_NS),
    ("i", XSI_NS),
    ("z", SERIALIZATION_NS),
)

# Input ceilings are derived from the largest supported committed inputs plus
# generous headroom: the largest spec fixture is about 81 KiB and the largest
# generated model is about 1.7 MiB. Bounds are enforced before parsing so an
# oversized document cannot be expanded in memory first.
MAX_SPEC_BYTES = 8 * 1024 * 1024
MAX_MODEL_BYTES = 64 * 1024 * 1024

# TMT stores ThreatInstance.Id as a signed 32-bit integer and treats 0 as the
# unassigned sentinel, so derived identifiers occupy the closed interval
# [1, 2147483647]. The identifier is a fixed standard-library digest over the
# UTF-8 bytes of the exact normalized source id, which keeps a threat's visible
# id constant across insertion, reordering, and separate processes.
THREAT_ID_DIGEST = "sha256"
THREAT_ID_MIN = 1
THREAT_ID_MAX = 2147483647

THREAT_MEMBER_ORDER = [
    "ChangedBy",
    "DrawingSurfaceGuid",
    "FlowGuid",
    "Id",
    "InteractionKey",
    "InteractionString",
    "ModifiedAt",
    "Priority",
    "Properties",
    "SourceGuid",
    "State",
    "StateInformation",
    "TargetGuid",
    "Title",
    "TypeId",
    "Upgraded",
    "UserThreatCategory",
    "UserThreatDescription",
    "UserThreatShortDescription",
    "Wide",
]
THREAT_PROPERTY_KEYS = [
    "Title",
    "UserThreatCategory",
    "UserThreatShortDescription",
    "UserThreatDescription",
    "InteractionString",
    "PossibleMitigations",
    "Priority",
    "SDLPhase",
]
STATE_ALIASES = {
    "Accepted (Outside Control)": "NotApplicable",
    "Accepted with Documentation": "NotApplicable",
    "AutoGenerated": "AutoGenerated",
    "Migrated": "Migrated",
    "Mitigated": "Mitigated",
    "Mitigated (Copilot Controls)": "Mitigated",
    "Mitigated by Design": "Mitigated",
    "Mitigated with Documentation": "Mitigated",
    "NeedsInvestigation": "NeedsInvestigation",
    "Needs-Investigation": "NeedsInvestigation",
    "NeedsMitigation": "NeedsMitigation",
    "NotApplicable": "NotApplicable",
    "Not-Applicable-False-Positive": "NotApplicable",
    "Open": "NeedsInvestigation",
    "Partially Mitigated": "NeedsMitigation",
    "Partially Mitigated with Documentation": "NeedsMitigation",
}
NULL_GUID = "00000000-0000-0000-0000-000000000000"


class UnsafeXmlError(ValueError):
    """Raised when XML input declares a DTD or entities, or cannot be parsed."""


class InputTooLargeError(ValueError):
    """Raised when an untrusted input exceeds its documented size ceiling."""


# ElementTree resolves prefixes through the process-global ``ET._namespace_map``
# and offers no per-document alternative, so every TM7 write passes through this
# single boundary. The lock serializes concurrent writers so one cannot observe
# another's prefixes, and the snapshot restores the caller's registry on every
# exit path including an exception.
_NAMESPACE_REGISTRY_LOCK = threading.Lock()


@contextmanager
def tm7_serialization_namespaces(root: ET.Element) -> Iterator[None]:
    """Bind TM7 prefixes and the root xsd attribute for one serialization.

    Args:
        root: Model root element that carries the ``xmlns:c`` schema binding.

    Yields:
        None, while the TM7 prefix bindings are active.
    """
    with _NAMESPACE_REGISTRY_LOCK:
        previous_map = ET._namespace_map.copy()
        try:
            for prefix, uri in TM7_NAMESPACE_PREFIXES:
                ET.register_namespace(prefix, uri)
            root.set("xmlns:c", XSD_NS)
            yield
        finally:
            root.attrib.pop("xmlns:c", None)
            ET._namespace_map.clear()
            ET._namespace_map.update(previous_map)


def read_bounded_bytes(path: Path, max_bytes: int) -> bytes:
    """Read a file only when it fits inside its documented size ceiling.

    The size is checked before any content is loaded, so an oversized document
    is rejected without being expanded in memory.

    Args:
        path: File to read.
        max_bytes: Maximum accepted size in bytes.

    Returns:
        Raw file bytes.

    Raises:
        InputTooLargeError: If the file exceeds ``max_bytes``.
        OSError: If the file cannot be inspected or read.
    """
    size = path.stat().st_size
    if size > max_bytes:
        raise InputTooLargeError(
            f"{path} is {size} bytes, which exceeds the {max_bytes} byte limit"
        )
    return path.read_bytes()


def contains_doctype_or_entity(data: bytes) -> bool:
    """Return True when the payload declares a DTD or entity in any encoding.

    A byte-level scan for ``<!DOCTYPE`` only matches UTF-8. The same document
    encoded as UTF-16 interleaves NUL bytes, so the marker never appears and
    the check silently passes. The payload is therefore decoded through the
    encodings XML permits before scanning.
    """
    for encoding in ("utf-8", "utf-16", "utf-16-le", "utf-16-be", "utf-32"):
        try:
            text = data.decode(encoding)
        except (UnicodeDecodeError, LookupError, ValueError):
            continue
        upper = text.upper()
        if "<!DOCTYPE" in upper or "<!ENTITY" in upper:
            return True
    return False


def parse_hardened_xml_bytes(data: bytes) -> ET.Element:
    """Parse XML bytes under one fail-closed policy shared by every reader.

    Named for its byte input so it cannot be confused with
    ``generate_tm7.parse_hardened_xml``, which takes a path and returns a
    parsed model payload rather than an element.

    Every entry point rejects DTD and entity declarations in any supported
    encoding, parses through defusedxml, which is a required runtime
    dependency, and converts parser and defusedxml exceptions alike into
    ``UnsafeXmlError``. Callers translate that into their own domain error so
    an unsafe document never surfaces as an unhandled traceback.

    Args:
        data: Raw XML bytes from an untrusted source.

    Returns:
        Parsed root element.

    Raises:
        UnsafeXmlError: If the payload declares a DTD or entity, or cannot be
            parsed safely.
    """
    if contains_doctype_or_entity(data):
        raise UnsafeXmlError("Refusing to parse XML with DTD or entity declarations")
    try:
        return DefusedET.fromstring(data)
    except ET.ParseError as exc:
        raise UnsafeXmlError(f"Unable to parse XML: {exc}") from exc
    except Exception as exc:  # defusedxml raises its own DTD/entity errors
        raise UnsafeXmlError(f"Refusing to parse XML: {exc}") from exc


class ThreatContractError(ValueError):
    """Raised when a threat-instance payload violates the TM7 contract."""


def _normalize_text(value: Any) -> str:
    return "" if value is None else str(value).strip()


def _coerce_list(value: Any) -> list[Any]:
    if value is None:
        return []
    return value if isinstance(value, list) else [value]


def _local_name(tag: str) -> str:
    return tag.rsplit("}", 1)[-1]


def _node_text(node: ET.Element | None) -> str:
    return "" if node is None or node.text is None else node.text.strip()


def _optional_node_text(node: ET.Element | None) -> str | None:
    return None if node is None or node.text is None else node.text.strip()


def _flow_reference(value: Any) -> str:
    if isinstance(value, dict):
        return _normalize_text(value.get("id"))
    return _normalize_text(value)


def normalize_state(value: Any) -> str:
    """Translate a source state into a TMT 7.3.51110.1 enum name."""
    state = _normalize_text(value) or "Open"
    try:
        return STATE_ALIASES[state]
    except KeyError as exc:
        raise ThreatContractError(f"unsupported threat state {state}") from exc


def build_entry_key(
    type_id: str,
    source_guid: str,
    flow_guid: str,
    target_guid: str,
) -> str:
    """Build the fixture-confirmed threat dictionary key."""
    return f"{type_id}{source_guid}{flow_guid}{target_guid}"


def build_interaction_key(
    source_guid: str,
    flow_guid: str,
    target_guid: str,
) -> str:
    """Build the fixture-confirmed colon-delimited interaction key."""
    return f"{source_guid}:{flow_guid}:{target_guid}"


def build_threat_instance_properties(
    threat: dict[str, Any],
) -> list[tuple[str, str]]:
    """Build the eight fixture-confirmed threat property entries."""
    supplied = threat.get("properties")
    supplied = supplied if isinstance(supplied, dict) else {}
    return [
        ("Title", _normalize_text(supplied.get("Title") or threat.get("title"))),
        (
            "UserThreatCategory",
            _normalize_text(
                supplied.get("UserThreatCategory") or threat.get("category")
            ),
        ),
        (
            "UserThreatShortDescription",
            _normalize_text(
                supplied.get("UserThreatShortDescription") or threat.get("description")
            ),
        ),
        (
            "UserThreatDescription",
            _normalize_text(
                supplied.get("UserThreatDescription") or threat.get("description")
            ),
        ),
        (
            "InteractionString",
            _normalize_text(
                supplied.get("InteractionString")
                or threat.get("interaction_string")
                or threat.get("interaction_ref")
            ),
        ),
        (
            "PossibleMitigations",
            _normalize_text(
                supplied.get("PossibleMitigations") or threat.get("mitigations")
            ),
        ),
        ("Priority", _normalize_text(supplied.get("Priority")) or "High"),
        ("SDLPhase", _normalize_text(supplied.get("SDLPhase")) or "Design"),
    ]


def _make_guid(identifier: str) -> str:
    """Create a deterministic GUID from a stable identifier.

    Lives here alongside the other identity builders because both generators
    mint GUIDs that must agree for the same identifier. Two independent copies
    of a deterministic identity function is a correctness hazard, not just
    duplication: a divergence would silently renumber every element.
    """
    return str(uuid.uuid5(uuid.NAMESPACE_URL, identifier))


def build_custom_threat_type_id(source_id: str) -> str:
    """Build the deterministic ThreatType identifier for a custom spec threat.

    Both TM7 producers emit the same DataContract, so slug derivation must be
    identical. Runs of characters outside ``a-z0-9`` collapse to a single
    hyphen, which keeps identifiers stable when punctuation abuts a hyphen.
    """
    normalized = _normalize_text(source_id) or "spec-threat"
    slug = (re.sub(r"[^a-z0-9]+", "-", normalized.lower()).strip("-") or "spec-threat")[
        :24
    ]
    digest = hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16].upper()
    return f"THC-{slug}-{digest}"


def build_mitigation_text(spec: dict[str, Any], threat: dict[str, Any]) -> str:
    """Resolve a threat's declared mitigation identifiers into display text."""
    mitigations = {
        _normalize_text(item.get("id")): item
        for item in _coerce_list(spec.get("mitigations"))
        if isinstance(item, dict) and item.get("id")
    }
    values: list[str] = []
    for mitigation_id in _coerce_list(threat.get("mitigation_ids")):
        mitigation = mitigations.get(_normalize_text(mitigation_id))
        if not isinstance(mitigation, dict):
            continue
        value = _normalize_text(mitigation.get("description") or mitigation.get("name"))
        if value:
            values.append(value)
    return "; ".join(dict.fromkeys(values))


def _surface_metadata(spec: dict[str, Any]) -> list[dict[str, Any]]:
    representations = spec.get("representations")
    if not isinstance(representations, dict):
        return []
    surfaces: list[dict[str, Any]] = []
    for group in (
        "context_diagrams",
        "functional_scenarios",
        "operational_views",
    ):
        for surface in _coerce_list(representations.get(group)):
            if not isinstance(surface, dict):
                continue
            members = surface.get("components") or surface.get("elements") or []
            surfaces.append(
                {
                    "id": _normalize_text(surface.get("id")),
                    "members": {
                        _normalize_text(member.get("id"))
                        for member in members
                        if isinstance(member, dict) and member.get("id")
                    },
                    "flow_ids": {
                        _flow_reference(flow_id)
                        for flow_id in _coerce_list(surface.get("flows"))
                        if _flow_reference(flow_id)
                    },
                }
            )
    return surfaces


def _placement_override(threat: dict[str, Any]) -> tuple[bool, str | None]:
    override = threat.get("placement_override")
    if override is None:
        return False, None
    if not isinstance(override, dict):
        return False, None
    rationale = _normalize_text(override.get("rationale"))
    reviewed = override.get("reviewed")
    if reviewed is True and rationale:
        return True, rationale
    return False, rationale or None


def _surface_matchers(
    spec: dict[str, Any],
    interaction_ref: str,
    source_ref: str,
    target_ref: str,
    threat_target: str,
) -> list[dict[str, Any]]:
    endpoint_targets = {source_ref, target_ref}
    return [
        surface
        for surface in _surface_metadata(spec)
        if interaction_ref in surface["flow_ids"]
        and endpoint_targets.issubset(surface["members"])
        and threat_target in surface["members"]
    ]


def collect_mapping_failures(
    spec: dict[str, Any],
    *,
    expected_count: int | None = None,
) -> list[str]:
    """Collect deterministic source-threat-to-flow topology failures."""
    if not isinstance(spec, dict):
        return ["spec root must be a mapping"]

    raw_flows = _coerce_list(spec.get("data_flows"))
    flow_counts = Counter(
        _normalize_text(flow.get("id"))
        for flow in raw_flows
        if isinstance(flow, dict) and flow.get("id")
    )
    flows = {
        _normalize_text(flow.get("id")): flow
        for flow in raw_flows
        if isinstance(flow, dict) and flow.get("id")
    }
    failures: list[tuple[str, str]] = []
    seen_ids: set[str] = set()

    for ordinal, threat in enumerate(_coerce_list(spec.get("threats")), start=1):
        if not isinstance(threat, dict):
            failures.append((f"~{ordinal:04d}", "threat entry is not a mapping"))
            continue
        threat_id = _normalize_text(threat.get("id"))
        sort_id = threat_id or f"~{ordinal:04d}"
        if not threat_id:
            failures.append((sort_id, "missing threat id"))
            continue
        if threat_id in seen_ids:
            failures.append((sort_id, f"{threat_id}: duplicate threat id"))
            continue
        seen_ids.add(threat_id)

        interaction_ref = _normalize_text(threat.get("interaction_ref"))
        if not interaction_ref:
            failures.append((sort_id, f"{threat_id}: missing interaction_ref"))
            continue
        if flow_counts[interaction_ref] > 1:
            failures.append(
                (sort_id, f"{threat_id}: ambiguous interaction_ref {interaction_ref}")
            )
            continue
        flow = flows.get(interaction_ref)
        if flow is None:
            failures.append(
                (sort_id, f"{threat_id}: unknown interaction_ref {interaction_ref}")
            )
            continue

        source_ref = _normalize_text(flow.get("source_ref"))
        target_ref = _normalize_text(flow.get("target_ref"))
        threat_target = _normalize_text(threat.get("target_ref"))
        if not source_ref or not target_ref:
            failures.append(
                (sort_id, f"{threat_id}: null-linked interaction_ref {interaction_ref}")
            )
            continue
        if not threat_target:
            failures.append((sort_id, f"{threat_id}: missing target_ref"))
            continue

        endpoint_targets = {source_ref, target_ref}
        if threat_target not in endpoint_targets:
            override_is_valid, _ = _placement_override(threat)
            if not override_is_valid:
                failures.append(
                    (
                        sort_id,
                        f"{threat_id}: semantic target {threat_target} is not an "
                        f"endpoint of interaction_ref {interaction_ref} and "
                        "placement_override is required",
                    )
                )
                continue

        matching_surfaces = _surface_matchers(
            spec,
            interaction_ref,
            source_ref,
            target_ref,
            threat_target,
        )
        if len(matching_surfaces) != 1:
            failures.append(
                (
                    sort_id,
                    f"{threat_id}: interaction_ref {interaction_ref} and target_ref "
                    f"{threat_target} resolve to {len(matching_surfaces)} surfaces",
                )
            )

    if expected_count is not None and len(seen_ids) != expected_count:
        failures.append(
            (
                "~~count",
                f"expected {expected_count} unique threat ids, found {len(seen_ids)}",
            )
        )
    return [message for _, message in sorted(failures)]


def collect_gap_citation_failures(
    spec: dict[str, Any],
    registers: dict[str, set[str]],
) -> list[str]:
    """Collect gap citations that do not resolve against a per-skill register.

    Gap ids are scoped per skill and may repeat across skills, so ``G-INF-1``
    names a different weakness in each file that declares it. The citable
    identity is therefore the ``(source_skill, gap_id)`` pair, and a citation
    that omits ``source_skill`` is unresolvable rather than merely incomplete.

    ``registers`` maps a skill name to the gap ids that skill declares.
    """
    if not isinstance(spec, dict):
        return ["spec root must be a mapping"]

    failures: list[tuple[str, str]] = []
    for ordinal, threat in enumerate(_coerce_list(spec.get("threats")), start=1):
        if not isinstance(threat, dict):
            continue
        threat_id = _normalize_text(threat.get("id"))
        sort_id = threat_id or f"~{ordinal:04d}"
        properties = threat.get("properties")
        if not isinstance(properties, dict):
            continue
        gap_ids = [
            _normalize_text(gap) for gap in _coerce_list(properties.get("gap_ids"))
        ]
        gap_ids = [gap for gap in gap_ids if gap]
        if not gap_ids:
            continue

        source_skill = _normalize_text(properties.get("source_skill"))
        if not source_skill:
            failures.append(
                (
                    sort_id,
                    f"{threat_id}: cites {', '.join(sorted(gap_ids))} without a "
                    "source_skill, so the gap ids cannot be resolved",
                )
            )
            continue

        register = registers.get(source_skill)
        if register is None:
            failures.append(
                (
                    sort_id,
                    f"{threat_id}: source_skill {source_skill} has no gap register",
                )
            )
            continue

        for gap_id in sorted(set(gap_ids)):
            if gap_id not in register:
                failures.append(
                    (
                        f"{sort_id}:{gap_id}",
                        f"{threat_id}: {gap_id} is not declared by {source_skill}",
                    )
                )
    return [message for _, message in sorted(failures)]


def _coerce_authored_base_index(authored_base: dict[str, Any]) -> dict[str, Any]:
    """Normalize the one supported authored-base index shape.

    ``populate_tm7_threats._build_authored_base_index`` is the single producer.
    Each surface entry carries the declared surface ``id`` it was resolved to,
    the authored ``guid``, and an ``elements`` map keyed by declared element id.
    """
    return {
        "connectors": authored_base.get("connectors") or {},
        "surfaces": authored_base.get("surfaces") or [],
    }


def _index_authored_surfaces(
    authored_surfaces: Any,
) -> tuple[dict[str, dict[str, Any]], list[str]]:
    """Index authored surfaces by declared identity and report duplicates.

    Authored surfaces that resolve to no declared surface carry an empty id and
    are ignored; a base model may legitimately contain extra diagrams.
    """
    by_id: dict[str, dict[str, Any]] = {}
    duplicates: set[str] = set()
    for entry in _coerce_list(authored_surfaces):
        if not isinstance(entry, dict):
            continue
        surface_id = _normalize_text(entry.get("id"))
        if not surface_id:
            continue
        if surface_id in by_id:
            duplicates.add(surface_id)
            continue
        by_id[surface_id] = entry
    failures = [
        f"authored-base surface {surface_id} is declared more than once"
        for surface_id in sorted(duplicates)
    ]
    return by_id, failures


def reconcile_authored_base(
    spec: dict[str, Any],
    authored_base: dict[str, Any] | None,
) -> list[str]:
    """Reconcile a spec against an authored base model for connector placement."""
    if not isinstance(spec, dict):
        return ["spec root must be a mapping"]
    if authored_base is None:
        return []
    if not isinstance(authored_base, dict):
        return ["authored base must be an index mapping"]

    index = _coerce_authored_base_index(authored_base)
    connectors = index.get("connectors") or {}
    authored_surfaces, failures = _index_authored_surfaces(index.get("surfaces"))
    declared_surfaces = _surface_metadata(spec)
    flows = {
        _normalize_text(flow.get("id")): flow
        for flow in _coerce_list(spec.get("data_flows"))
        if isinstance(flow, dict) and flow.get("id")
    }

    for threat in _coerce_list(spec.get("threats")):
        if not isinstance(threat, dict):
            continue
        threat_id = _normalize_text(threat.get("id")) or "~threat"
        interaction_ref = _normalize_text(threat.get("interaction_ref"))
        if not interaction_ref:
            continue
        flow = flows.get(interaction_ref)
        if flow is None:
            continue
        source_ref = _normalize_text(flow.get("source_ref"))
        target_ref = _normalize_text(flow.get("target_ref"))
        expected_matches = [
            surface
            for surface in declared_surfaces
            if interaction_ref in surface["flow_ids"]
            and {source_ref, target_ref}.issubset(surface["members"])
        ]
        if len(expected_matches) != 1:
            failures.append(
                f"{threat_id}: authored-base expected surface for "
                f"{interaction_ref} is ambiguous"
            )
            continue
        expected_surface_id = _normalize_text(expected_matches[0].get("id"))
        authored_surface = authored_surfaces.get(expected_surface_id)
        if authored_surface is None:
            failures.append(
                f"{threat_id}: authored-base surface "
                f"{expected_surface_id or '~unidentified'} for {interaction_ref} "
                "is absent"
            )
            continue
        expected_surface_guid = _normalize_text(authored_surface.get("guid"))

        raw_connectors = connectors.get(interaction_ref)
        candidates = (
            raw_connectors
            if isinstance(raw_connectors, list)
            else [raw_connectors]
            if isinstance(raw_connectors, dict)
            else []
        )
        connector = next(
            (
                candidate
                for candidate in candidates
                if _normalize_text(candidate.get("drawing_surface_guid"))
                == expected_surface_guid
            ),
            None,
        )
        if connector is None:
            if candidates:
                failures.append(
                    f"{threat_id}: authored-base surface mismatch for {interaction_ref}"
                )
            else:
                failures.append(
                    f"{threat_id}: authored-base connector {interaction_ref} is absent"
                )
            continue
        guid_values = [
            expected_surface_guid,
            _normalize_text(connector.get("flow_guid")),
            _normalize_text(connector.get("source_guid")),
            _normalize_text(connector.get("target_guid")),
        ]
        if any(not value or value == NULL_GUID for value in guid_values):
            failures.append(
                f"{threat_id}: authored-base connector {interaction_ref} has null GUIDs"
            )
            continue

        surface_elements = authored_surface.get("elements") or {}
        endpoint_guid_map = {
            source_ref: surface_elements.get(source_ref),
            target_ref: surface_elements.get(target_ref),
        }
        if not all(endpoint_guid_map.values()):
            failures.append(
                f"{threat_id}: authored-base connector {interaction_ref} "
                "endpoint identity is unavailable"
            )
            continue
        if (
            connector.get("source_guid") != endpoint_guid_map[source_ref]
            or connector.get("target_guid") != endpoint_guid_map[target_ref]
        ):
            failures.append(
                f"{threat_id}: authored-base connector {interaction_ref} "
                "endpoint identity mismatch"
            )

    return failures


def validate_serializable_threat_node(threat_node: ET.Element) -> None:
    """Validate the exact member ordering used by the TM7 threat DataContract."""
    member_order = [_local_name(child.tag) for child in threat_node]
    if member_order != THREAT_MEMBER_ORDER:
        raise ThreatContractError(
            f"threat member order {member_order} does not match the contract"
        )


def validate_threat_instance(
    threat: dict[str, Any],
    *,
    type_ids: set[str] | None = None,
) -> None:
    """Validate one fully resolved threat-instance payload."""
    if not isinstance(threat, dict):
        raise ThreatContractError("threat payload must be a mapping")
    threat_id = threat.get("id")
    if (
        not isinstance(threat_id, int)
        or isinstance(threat_id, bool)
        or not THREAT_ID_MIN <= threat_id <= THREAT_ID_MAX
    ):
        raise ThreatContractError(
            f"threat id must be an integer in [{THREAT_ID_MIN}, {THREAT_ID_MAX}]"
        )

    if not _normalize_text(threat.get("interaction_ref")):
        raise ThreatContractError("threat interaction_ref is required")

    type_id = _normalize_text(threat.get("type_id"))
    if not type_id:
        raise ThreatContractError("threat type_id is required")
    if type_ids is not None and type_id not in type_ids:
        raise ThreatContractError(
            f"threat type_id {type_id} is not embedded in KnowledgeBase"
        )

    guids = [
        _normalize_text(threat.get(name))
        for name in (
            "drawing_surface_guid",
            "source_guid",
            "flow_guid",
            "target_guid",
        )
    ]
    if any(not guid or guid == NULL_GUID for guid in guids):
        raise ThreatContractError("threat guid members must be non-null")

    expected_interaction = build_interaction_key(*guids[1:])
    if _normalize_text(threat.get("interaction_key")) != expected_interaction:
        raise ThreatContractError("interaction_key does not match the GUID tuple")
    expected_dictionary = build_entry_key(type_id, *guids[1:])
    if _normalize_text(threat.get("dictionary_key")) != expected_dictionary:
        raise ThreatContractError("dictionary_key does not match the threat contract")
    normalize_state(threat.get("state"))


def derive_threat_numeric_id(source_id: str) -> int:
    """Derive the stable TMT-visible numeric id for a normalized source id."""
    if not source_id:
        raise ThreatContractError("source threat id is required to derive an id")
    digest = hashlib.new(THREAT_ID_DIGEST, source_id.encode("utf-8")).digest()
    span = THREAT_ID_MAX - THREAT_ID_MIN + 1
    return THREAT_ID_MIN + int.from_bytes(digest, "big") % span


def prepare_threat_instances(
    threats: list[dict[str, Any]],
    *,
    type_ids: set[str] | None = None,
) -> list[dict[str, Any]]:
    """Assign stable numeric IDs and validate collection-wide uniqueness."""
    prepared: list[dict[str, Any]] = []
    seen_source_ids: set[str] = set()
    seen_keys: set[str] = set()
    numeric_owners: dict[int, str] = {}
    for threat in sorted(
        threats, key=lambda item: _normalize_text(item.get("source_id"))
    ):
        source_id = _normalize_text(threat.get("source_id"))
        if not source_id or source_id in seen_source_ids:
            raise ThreatContractError(
                f"duplicate or missing source threat id {source_id}"
            )
        seen_source_ids.add(source_id)
        numeric_id = derive_threat_numeric_id(source_id)
        collided_with = numeric_owners.get(numeric_id)
        if collided_with is not None:
            raise ThreatContractError(
                f"threat id collision: {collided_with} and {source_id} both "
                f"derive numeric id {numeric_id}"
            )
        numeric_owners[numeric_id] = source_id
        resolved = {
            **threat,
            "id": numeric_id,
            "state": normalize_state(threat.get("state")),
        }
        validate_threat_instance(resolved, type_ids=type_ids)
        dictionary_key = _normalize_text(resolved.get("dictionary_key"))
        if dictionary_key in seen_keys:
            raise ThreatContractError(f"duplicate dictionary key for {source_id}")
        seen_keys.add(dictionary_key)
        prepared.append(resolved)
    return prepared


def _add_value(parent: ET.Element, name: str, value: str | None) -> ET.Element:
    node = ET.SubElement(parent, f"{{{KNOWLEDGE_NS}}}{name}")
    if value is None:
        node.set(f"{{{XSI_NS}}}nil", "true")
    else:
        node.text = value
    return node


def serialize_threat_instance(threat: dict[str, Any]) -> ET.Element:
    """Serialize one validated threat value in exact DataContract member order."""
    validate_threat_instance(threat)
    value = ET.Element(f"{{{ARRAYS_NS}}}Value")
    _add_value(value, "ChangedBy", None)
    _add_value(value, "DrawingSurfaceGuid", threat["drawing_surface_guid"])
    _add_value(value, "FlowGuid", threat["flow_guid"])
    _add_value(value, "Id", str(threat["id"]))
    _add_value(value, "InteractionKey", threat["interaction_key"])
    _add_value(value, "InteractionString", None)
    _add_value(value, "ModifiedAt", "0001-01-01T00:00:00")
    _add_value(value, "Priority", "High")
    properties = _add_value(value, "Properties", "")
    properties.text = None
    for property_key, property_value in build_threat_instance_properties(threat):
        item = ET.SubElement(properties, f"{{{ARRAYS_NS}}}KeyValueOfstringstring")
        ET.SubElement(item, f"{{{ARRAYS_NS}}}Key").text = property_key
        ET.SubElement(item, f"{{{ARRAYS_NS}}}Value").text = property_value
    _add_value(value, "SourceGuid", threat["source_guid"])
    _add_value(value, "State", threat["state"])
    _add_value(value, "StateInformation", None)
    _add_value(value, "TargetGuid", threat["target_guid"])
    _add_value(value, "Title", None)
    _add_value(value, "TypeId", threat["type_id"])
    _add_value(value, "Upgraded", "false")
    _add_value(value, "UserThreatCategory", None)
    _add_value(value, "UserThreatDescription", None)
    _add_value(value, "UserThreatShortDescription", None)
    _add_value(value, "Wide", "false")
    return value


def serialize_threat_instances(
    root: ET.Element,
    threats: list[dict[str, Any]],
    *,
    type_ids: set[str] | None = None,
) -> list[dict[str, Any]]:
    """Replace the root ThreatInstances member with validated entries."""
    prepared = prepare_threat_instances(threats, type_ids=type_ids)
    existing = next(
        (child for child in root if _local_name(child.tag) == "ThreatInstances"),
        None,
    )
    if existing is None:
        threat_instances = ET.Element("ThreatInstances")
        insert_at = next(
            (
                index
                for index, child in enumerate(root)
                if _local_name(child.tag) == "ThreatGenerationEnabled"
            ),
            len(root),
        )
        root.insert(insert_at, threat_instances)
    else:
        threat_instances = existing
        threat_instances.clear()

    for threat in prepared:
        item = ET.SubElement(
            threat_instances,
            f"{{{ARRAYS_NS}}}KeyValueOfstringThreatpc_P0_PhOB",
        )
        ET.SubElement(item, f"{{{ARRAYS_NS}}}Key").text = threat["dictionary_key"]
        item.append(serialize_threat_instance(threat))
    return prepared


def extract_serializable_threat(threat_node: ET.Element) -> dict[str, Any]:
    """Extract a serialized threat value for contract regression assertions."""
    validate_serializable_threat_node(threat_node)
    child_map = {_local_name(child.tag): child for child in threat_node}
    properties: dict[str, str] = {}
    properties_node = child_map.get("Properties")
    if properties_node is not None:
        for item in properties_node.findall("{*}KeyValueOfstringstring"):
            properties[_node_text(item.find("{*}Key"))] = _node_text(
                item.find("{*}Value")
            )
    content_fields = [
        "Title",
        "UserThreatCategory",
        "UserThreatShortDescription",
        "UserThreatDescription",
        "InteractionString",
        "StateInformation",
    ]
    return {
        "member_order": [_local_name(child.tag) for child in threat_node],
        "id": _node_text(child_map.get("Id")),
        "type_id": _node_text(child_map.get("TypeId")),
        "drawing_surface_guid": _node_text(child_map.get("DrawingSurfaceGuid")),
        "source_guid": _node_text(child_map.get("SourceGuid")),
        "flow_guid": _node_text(child_map.get("FlowGuid")),
        "target_guid": _node_text(child_map.get("TargetGuid")),
        "interaction_key": _node_text(child_map.get("InteractionKey")),
        "state": _node_text(child_map.get("State")),
        "properties": properties,
        "top_level_content": {
            field: _optional_node_text(child_map.get(field)) for field in content_fields
        },
    }


def validate_serialized_threat_entry(
    entry: ET.Element,
    *,
    type_ids: set[str] | None = None,
) -> dict[str, Any]:
    """Validate one fixture-shaped threat dictionary entry."""
    key_node = entry.find("{*}Key")
    value_node = entry.find("{*}Value")
    if key_node is None or value_node is None:
        raise ThreatContractError("threat entry must contain Key and Value")

    extracted = extract_serializable_threat(value_node)
    if extracted["member_order"] != THREAT_MEMBER_ORDER:
        raise ThreatContractError("threat members are not in DataContract order")

    threat_id = str(extracted["id"])
    if not threat_id.isdigit() or int(threat_id) <= 0:
        raise ThreatContractError("threat id must be a positive integer")

    type_id = str(extracted["type_id"])
    if not type_id:
        raise ThreatContractError("threat type_id is required")
    if type_ids is not None and type_id not in type_ids:
        raise ThreatContractError(
            f"threat type_id {type_id} is not embedded in KnowledgeBase"
        )

    guids = [
        str(extracted[name])
        for name in (
            "drawing_surface_guid",
            "source_guid",
            "flow_guid",
            "target_guid",
        )
    ]
    if any(not guid or guid == NULL_GUID for guid in guids):
        raise ThreatContractError("threat guid members must be non-null")

    expected_interaction = build_interaction_key(*guids[1:])
    if extracted["interaction_key"] != expected_interaction:
        raise ThreatContractError("interaction_key does not match the GUID tuple")
    expected_key = build_entry_key(type_id, *guids[1:])
    if _node_text(key_node) != expected_key:
        raise ThreatContractError("dictionary_key does not match the threat contract")

    if list(extracted["properties"]) != THREAT_PROPERTY_KEYS:
        raise ThreatContractError("threat properties do not match fixture order")
    if str(extracted["state"]) not in set(STATE_ALIASES.values()):
        raise ThreatContractError(f"unsupported threat state {extracted['state']}")
    if any(value is not None for value in extracted["top_level_content"].values()):
        raise ThreatContractError("top-level threat content fields must be nil")
    return extracted


def validate_serialized_threat_entries(
    threat_instances: ET.Element,
    *,
    type_ids: set[str] | None = None,
) -> list[dict[str, Any]]:
    """Validate all serialized entries and collection-wide uniqueness."""
    extracted_entries: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    seen_keys: set[str] = set()
    for entry in threat_instances.findall("{*}KeyValueOfstringThreatpc_P0_PhOB"):
        extracted = validate_serialized_threat_entry(entry, type_ids=type_ids)
        threat_id = str(extracted["id"])
        key = _node_text(entry.find("{*}Key"))
        if threat_id in seen_ids:
            raise ThreatContractError(f"duplicate numeric threat id {threat_id}")
        if key in seen_keys:
            raise ThreatContractError("duplicate threat dictionary key")
        seen_ids.add(threat_id)
        seen_keys.add(key)
        extracted_entries.append(extracted)
    return extracted_entries