Scope boundary

Owns the authoritative identity, version, owner, lifecycle state, and security-relevant attributes of AI components, models, datasets, infrastructure, and accelerators.

AID-M-001.002: AID-M-001.001 inventories the assets that exist; AID-M-001.002 maps the dependency edges and release composition that connect those assets. An asset record can be complete without proving its upstream and downstream relationships, while a dependency edge must resolve to inventoried endpoints.

Implementation guidance

• AID-M-001.001-G001: Establish and maintain a dynamic, up-to-date registry of AI models and datasets with security-relevant metadata.
• AID-M-001.001-G002: Include specialized AI accelerators (GPUs, TPUs, NPUs, FPGAs) and their firmware versions in the AI asset inventory.
• AID-M-001.001-G003: Assign clear ownership and accountability for each inventoried AI asset.
• AID-M-001.001-G004: Integrate AI asset inventory with broader IT asset management and configuration management databases (CMDBs).

AID-M-001.001-G001

Concept:

This step is specifically about the model-and-dataset registry layer, not the whole software or infrastructure inventory. The goal is to maintain a living operational index for every governed model version and the datasets that trained or validated it, including owner, deployment alias, digest, and approval state. Registry aliases and tags are mutable operational metadata; they are useful discovery pointers but are never durable authoritative evidence.

Step 1: Stand up a persistent tracking and registry service

Use a registry that can store model versions, dataset references, tags, and correlated run metadata. The example below uses MLflow with a persistent metadata store.

# 1. Install MLflow and common ML dependencies
# requirements.lock is reviewed, hash-pinned, and bound to the signed inventory-run manifest.
python -m pip install --require-hashes --no-deps -r requirements.lock

# 2. Start MLflow with persistent metadata and artifact storage
mlflow server   --backend-store-uri sqlite:///mlflow.db   --default-artifact-root ./mlruns   --host 127.0.0.1   --port 5000

# In production, replace local paths with managed storage:
# - Postgres/MySQL for backend-store-uri
# - S3/GCS/Azure Blob for default-artifact-root

Step 2: Log the dataset reference and model version in the same training run

Capture dataset lineage, model parameters, owner metadata, and the registered model version in one correlated operational run record.

from __future__ import annotations

# File: train_model.py
import hashlib
import json
import math
import os
import stat
import subprocess
import tempfile
from pathlib import Path
from urllib.parse import urlparse

import mlflow
import mlflow.data
import mlflow.sklearn
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return "sha256:" + digest.hexdigest()


def strict_json_bytes(raw: bytes, label: str) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError(f"non-finite JSON constant in {label}: {value}")

    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_object,
            parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
        raise RuntimeError(f"{label} is not strict UTF-8 JSON") from error


def read_regular_bytes(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    with os.fdopen(descriptor, "rb") as handle:
        status = os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode) or status.st_size == 0:
            raise RuntimeError(f"missing, empty, or unsafe signed input: {path}")
        return handle.read()


def write_private_snapshot(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags, 0o400)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(raw)
        handle.flush()
        os.fsync(handle.fileno())


def capture_verified_bundle(
    payload_path: Path,
    bundle_path: Path,
    verification_key: str,
    timeout_seconds: float,
) -> bytes:
    payload = read_regular_bytes(payload_path)
    bundle = read_regular_bytes(bundle_path)
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        snapshot_root = Path(directory)
        snapshot_root.chmod(0o700)
        payload_snapshot = snapshot_root / "payload"
        bundle_snapshot = snapshot_root / "bundle.json"
        write_private_snapshot(payload_snapshot, payload)
        write_private_snapshot(bundle_snapshot, bundle)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", verification_key,
                "--bundle", str(bundle_snapshot), str(payload_snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        if read_regular_bytes(payload_snapshot) != payload or read_regular_bytes(bundle_snapshot) != bundle:
            raise RuntimeError("private signed-input snapshot changed during verification")
    return payload


def load_signed_training_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["M001001_TRAINING_POLICY_PATH"])
    bundle_path = Path(os.environ["M001001_TRAINING_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = capture_verified_bundle(
        policy_path,
        bundle_path,
        os.environ["M001001_TRAINING_POLICY_VERIFY_KEY"],
        verify_timeout,
    )
    policy = strict_json_bytes(raw, "signed training policy")
    required = {
        "schema_version", "policy_version", "mlflow_tracking_uri",
        "mlflow_experiment", "training_data_path", "training_data_source",
        "training_data_sha256", "dataset_name", "label_column",
        "test_fraction", "random_seed", "n_estimators",
        "registered_model_name",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed training-policy fields differ")
    if (
        policy["schema_version"] != "aidefend.training-run-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signed training-policy identity is invalid")
    for field in (
        "mlflow_tracking_uri", "mlflow_experiment", "training_data_path",
        "training_data_source", "dataset_name", "label_column",
        "registered_model_name",
    ):
        if not isinstance(policy[field], str) or not policy[field].strip():
            raise RuntimeError("signed training-policy string is invalid: " + field)
    expected_data_sha256 = policy["training_data_sha256"]
    if (
        not isinstance(expected_data_sha256, str)
        or len(expected_data_sha256) != 71
        or not expected_data_sha256.startswith("sha256:")
        or set(expected_data_sha256[7:]) - set("0123456789abcdef")
    ):
        raise RuntimeError("signed training-data digest is invalid")
    parsed_tracking = urlparse(policy["mlflow_tracking_uri"])
    if parsed_tracking.scheme not in {"https", "file"}:
        raise RuntimeError("MLflow tracking URI must be HTTPS or a protected file store")
    test_fraction = policy["test_fraction"]
    if (
        isinstance(test_fraction, bool)
        or not isinstance(test_fraction, (int, float))
        or not math.isfinite(float(test_fraction))
        or not 0 < float(test_fraction) < 1
    ):
        raise RuntimeError("signed test fraction is invalid")
    for field in ("random_seed", "n_estimators"):
        value = policy[field]
        minimum = 0 if field == "random_seed" else 1
        if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
            raise RuntimeError("signed training hyperparameter is invalid: " + field)
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


POLICY, POLICY_SHA256 = load_signed_training_policy()
training_path = Path(POLICY["training_data_path"])
if not training_path.is_file() or training_path.is_symlink():
    raise RuntimeError("policy-bound training dataset is absent or unsafe")
training_sha256 = sha256_file(training_path)
if training_sha256 != POLICY["training_data_sha256"]:
    raise RuntimeError("training dataset differs from signed policy")

mlflow.set_tracking_uri(POLICY["mlflow_tracking_uri"])
mlflow.set_experiment(POLICY["mlflow_experiment"])
training_df = pd.read_parquet(training_path)
label_column = POLICY["label_column"]
if label_column not in training_df or training_df.empty:
    raise RuntimeError("training population or signed label column is unavailable")
dataset = mlflow.data.from_pandas(
    training_df,
    source=POLICY["training_data_source"],
    name=POLICY["dataset_name"],
)
X = training_df.drop(columns=[label_column])
y = training_df[label_column]
X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size=float(POLICY["test_fraction"]),
    random_state=POLICY["random_seed"],
)
model = RandomForestClassifier(
    n_estimators=POLICY["n_estimators"],
    random_state=POLICY["random_seed"],
)
model.fit(X_train, y_train)

with mlflow.start_run() as run:
    mlflow.log_input(dataset, context="training")
    mlflow.log_params({
        "n_estimators": POLICY["n_estimators"],
        "random_seed": POLICY["random_seed"],
        "test_fraction": POLICY["test_fraction"],
        "training_policy_version": POLICY["policy_version"],
        "training_policy_sha256": POLICY_SHA256,
        "training_data_sha256": training_sha256,
    })
    mlflow.set_tags({
        "aidefend.control": "AID-M-001.001",
        "aidefend.policy_version": POLICY["policy_version"],
        "aidefend.policy_sha256": POLICY_SHA256,
        "aidefend.training_data_sha256": training_sha256,
    })
    mlflow.log_metric("validation_accuracy", model.score(X_test, y_test))
    mlflow.log_dict(
        {
            "schema_version": "aidefend.training-run-binding.v1",
            "run_id": run.info.run_id,
            "policy_version": POLICY["policy_version"],
            "policy_sha256": POLICY_SHA256,
            "training_data_sha256": training_sha256,
            "training_row_count": len(training_df),
            "training_split_count": len(X_train),
            "validation_split_count": len(X_test),
        },
        "evidence/training_run_binding.json",
    )
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name=POLICY["registered_model_name"],
    )

Step 3: Point the deployment alias only at a governed version

After validation, set operational tags and aliases that make the version queryable during deployment, incident response, and rollback. The subsequent evidence-export guidance snapshots and signs their observed values outside MLflow.

from __future__ import annotations

# File: registry/promote_model.py

from mlflow.tracking import MlflowClient

client = MlflowClient(tracking_uri="http://127.0.0.1:5000")
model_name = "credit-card-fraud-rfc"
model_version = "17"

client.set_model_version_tag(model_name, model_version, "owner_team", "fraud_analytics_team")
client.set_model_version_tag(model_name, model_version, "security_status", "approved")
client.set_model_version_tag(model_name, model_version, "training_dataset", "credit-card-fraud-sanitized-v3")
client.set_model_version_tag(model_name, model_version, "change_ticket", "AISEC-417")
client.set_registered_model_alias(model_name, "production", model_version)
client.set_registered_model_alias(model_name, "last-approved", model_version)

Action: Use the registry as a mutable operational index only. Promotion is incomplete until an externally credentialed exporter has captured the alias-resolved version and required tags in a signed, append-only evidence snapshot; the next guidance provides that hard gate.

AID-M-001.001-G002

Concept:

Cloud instance-type declarations show expected capacity, not the accelerator, driver, or firmware actually present on a node. Collect measurements from the host/device boundary with vendor tooling and bind them to stable node, host, and device identities. The example below measures NVIDIA GPU UUID, PCI address, driver, and VBIOS; equivalent signed collectors are required for TPU, NPU, or FPGA fleets.

Step 1: Collect measured device state

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
ACCELERATOR_COLLECTOR_TIMEOUT_SECONDS = _aidefend_positive_float("M001001_ACCELERATOR_COLLECTOR_TIMEOUT_SECONDS")

# File: inventory/collect_nvidia_device_state.py

import csv
import hashlib
import json
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from io import StringIO
from pathlib import Path


OUTPUT = Path("out/accelerator-inventory.json")
NODE_NAME = os.environ.get("NODE_NAME", "").strip()
HOST_ID_PATH = Path("/etc/machine-id")
EXPECTED_PATH = Path(os.environ["SIGNED_EXPECTED_ACCELERATOR_INVENTORY"])
EXPECTED_VERIFIER_KEY = Path(os.environ["EXPECTED_ACCELERATOR_INVENTORY_VERIFIER_KEY"])


def validate_expected_inventory(value: object) -> dict:
    document_keys = {"schema_version", "inventory_version", "devices"}
    device_keys = {"node_name", "vendor", "device_uuid", "pci_bus_id"}
    if not isinstance(value, dict) or set(value) != document_keys:
        raise RuntimeError("signed expected accelerator inventory schema differs")
    if (
        value["schema_version"] != "aidefend.expected-accelerator-population.v1"
        or not isinstance(value["inventory_version"], str)
        or not value["inventory_version"].strip()
        or not isinstance(value["devices"], list)
        or not value["devices"]
    ):
        raise RuntimeError("signed expected accelerator inventory identity or population is invalid")
    identities = []
    for item in value["devices"]:
        if not isinstance(item, dict) or set(item) != device_keys:
            raise RuntimeError("signed expected accelerator device schema differs")
        if any(not isinstance(item[key], str) or not item[key].strip() for key in device_keys):
            raise RuntimeError("signed expected accelerator device field is empty")
        identities.append((item["node_name"], item["vendor"], item["device_uuid"], item["pci_bus_id"]))
    if len(identities) != len(set(identities)):
        raise RuntimeError("signed expected accelerator population contains duplicate devices")
    return value


def strict_json_bytes(raw: bytes, label: str) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError(f"non-finite JSON constant in {label}: {value}")

    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_object,
            parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
        raise RuntimeError(f"{label} is not strict UTF-8 JSON") from error


def read_regular_bytes(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    with os.fdopen(descriptor, "rb") as handle:
        status = os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode) or status.st_size == 0:
            raise RuntimeError(f"missing, empty, or unsafe signed input: {path}")
        return handle.read()


def write_private_snapshot(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags, 0o400)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(raw)
        handle.flush()
        os.fsync(handle.fileno())


def capture_verified_bundle(
    payload_path: Path,
    bundle_path: Path,
    verification_key: str,
    timeout_seconds: float,
) -> bytes:
    payload = read_regular_bytes(payload_path)
    bundle = read_regular_bytes(bundle_path)
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        snapshot_root = Path(directory)
        snapshot_root.chmod(0o700)
        payload_snapshot = snapshot_root / "payload"
        bundle_snapshot = snapshot_root / "bundle.json"
        write_private_snapshot(payload_snapshot, payload)
        write_private_snapshot(bundle_snapshot, bundle)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", verification_key,
                "--bundle", str(bundle_snapshot), str(payload_snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        if read_regular_bytes(payload_snapshot) != payload or read_regular_bytes(bundle_snapshot) != bundle:
            raise RuntimeError("private signed-input snapshot changed during verification")
    return payload
def main() -> None:
    reasons = []
    if not NODE_NAME:
        reasons.append("NODE_NAME is missing")
    if not HOST_ID_PATH.is_file():
        reasons.append("stable host identity is unavailable")

    expected_bundle = EXPECTED_PATH.with_name(EXPECTED_PATH.name + ".sigstore.json")
    expected_inventory = {}
    expected_population_sha256 = None
    try:
        expected_raw = capture_verified_bundle(
            EXPECTED_PATH,
            expected_bundle,
            str(EXPECTED_VERIFIER_KEY),
            ACCELERATOR_COLLECTOR_TIMEOUT_SECONDS,
        )
        expected_inventory = validate_expected_inventory(
            strict_json_bytes(expected_raw, "signed expected accelerator inventory")
        )
        expected_population_sha256 = hashlib.sha256(expected_raw).hexdigest()
    except (OSError, subprocess.SubprocessError, RuntimeError, ValueError) as error:
        reasons.append(f"expected accelerator population is unavailable or invalid: {error}")

    command = [
        "nvidia-smi",
        "--query-gpu=uuid,name,pci.bus_id,driver_version,vbios_version",
        "--format=csv,noheader,nounits",
    ]
    try:
        completed = subprocess.run(command, capture_output=True, text=True, check=False, timeout=ACCELERATOR_COLLECTOR_TIMEOUT_SECONDS)
    except (OSError, subprocess.TimeoutExpired) as error:
        completed = None
        reasons.append(f"vendor collector unavailable: {error}")

    devices = []
    if completed is not None and completed.returncode == 0:
        for row in csv.reader(StringIO(completed.stdout)):
            values = [value.strip() for value in row]
            if len(values) != 5 or any(not value or value.lower() in {"n/a", "unknown"} for value in values):
                reasons.append(f"incomplete measured device row: {values!r}")
                continue
            devices.append(dict(zip(("device_uuid", "model", "pci_bus_id", "driver_version", "vbios_version"), values)))
    elif completed is not None:
        reasons.append((completed.stderr or completed.stdout).strip() or "vendor collector failed")
    if not devices:
        reasons.append("no fully measured accelerator devices")

    expected_on_node = [
        item for item in expected_inventory.get("devices", []) if item.get("node_name") == NODE_NAME
    ]
    if not expected_on_node:
        reasons.append("signed expected population has no devices for this node")
    unsupported_vendors = sorted(
        {str(item.get("vendor")) for item in expected_on_node if item.get("vendor") != "NVIDIA"}
    )
    if unsupported_vendors:
        reasons.append(f"expected devices require other vendor adapters: {unsupported_vendors}")
    expected_ids = {
        (str(item.get("device_uuid")), str(item.get("pci_bus_id")))
        for item in expected_on_node
        if item.get("vendor") == "NVIDIA"
    }
    measured_ids = {(item["device_uuid"], item["pci_bus_id"]) for item in devices}
    if expected_ids != measured_ids:
        reasons.append("measured device UUID/PCI population differs from signed expected node population")

    host_id = HOST_ID_PATH.read_text(encoding="utf-8").strip() if HOST_ID_PATH.is_file() else ""
    report = {
        "schema_version": "aidefend.accelerator_inventory.v1",
        "measurement_status": "MEASURED" if not reasons else "UNMEASURED",
        "measured_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
        "node_name": NODE_NAME,
        "host_identity_sha256": hashlib.sha256(host_id.encode("utf-8")).hexdigest() if host_id else None,
        "collector": {"command": command, "vendor": "NVIDIA"},
        "expected_population_sha256": expected_population_sha256,
        "expected_device_ids": sorted(expected_ids),
        "devices": devices,
        "reasons": sorted(set(reasons)),
    }
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    if report["measurement_status"] != "MEASURED":
        raise SystemExit(f"Accelerator inventory incomplete; see {OUTPUT}")


if __name__ == "__main__":
    main()

Step 2: Reconcile and retain evidence

The collector returns only MEASURED or UNMEASURED for its exact node; it never declares fleet-level pass. Sign each canonical report with a host-collector identity and retain it in externally controlled append-only/WORM storage. An independent gate verifies the expected-population signer and digest, requires exactly one fresh trusted report for every expected node/device and vendor adapter, rejects extra measured devices, and checks approved driver/VBIOS baselines.

Action: PASS exists only at the independent exact-population gate. A declared instance type, one measured row, tool installation, command exit, or jq length is not proof of fleet driver/firmware state. Missing reports, vendor adapters, permissions, stable identities, firmware fields, expected devices, authoritative evidence, or any partial/extra population is INSUFFICIENT_DATA and fails inventory freshness.

AID-M-001.001-G003

Concept and ownership source

Keep the accountable owner in a signed, version-controlled ownership manifest keyed by the canonical asset digest, then publish the same owner identifier on the exact registry object. A registry tag is a searchable projection, not the authority by itself.

Step 1: Define the owner record

# File: inventory/asset_owners.yaml
schema_version: aidefend.asset-owners.v1
assets:
  sha256:4f5e0d2a9d47d0e6c2df7c515d0e0c14b9f0d5f865f60f3c8bff2d99d5b3a481:
    owner_id: fraud-analytics-team
    escalation: ml-fraud-oncall
    status: active

Generate the digest with the same canonical tree-digest function used below and review the manifest through the normal code-owner path. Sign the approved manifest before CI consumes it.

Step 2: Bind and read back the exact MLflow model version

# File: tools/bind_model_owner.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path

import mlflow.artifacts
import yaml
from mlflow import MlflowClient

OWNER_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{2,63}$")
DIGEST = re.compile(r"^sha256:[a-f0-9]{64}$")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def tree_sha256(path: Path) -> str:
    if path.is_symlink():
        raise ValueError("registry artifact root cannot be a symlink")
    files = [path] if path.is_file() else sorted(
        item for item in path.rglob("*") if item.is_file()
    )
    if not files:
        raise ValueError("registry artifact contains no files")
    digest = hashlib.sha256()
    for item in files:
        if item.is_symlink():
            raise ValueError(f"registry artifact contains a symlink: {item}")
        relative = item.name if path.is_file() else item.relative_to(path).as_posix()
        name = relative.encode("utf-8")
        size = item.stat().st_size
        digest.update(len(name).to_bytes(8, "big"))
        digest.update(name)
        digest.update(size.to_bytes(8, "big"))
        with item.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
    return "sha256:" + digest.hexdigest()


class UniqueKeySafeLoader(yaml.SafeLoader):
    pass


def _construct_unique_mapping(loader, node, deep=False):
    loader.flatten_mapping(node)
    mapping = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        try:
            duplicate = key in mapping
        except TypeError as error:
            raise ValueError("ownership manifest contains an unhashable mapping key") from error
        if duplicate:
            raise ValueError(f"ownership manifest contains duplicate key: {key!r}")
        mapping[key] = loader.construct_object(value_node, deep=deep)
    return mapping


UniqueKeySafeLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    _construct_unique_mapping,
)


def verified_manifest_snapshot(path: Path, signature: Path, key: Path) -> bytes:
    timeout_raw = os.environ["M001001_OWNER_MANIFEST_VERIFY_TIMEOUT_SECONDS"]
    try:
        timeout = float(timeout_raw)
    except ValueError as error:
        raise RuntimeError("owner-manifest verification timeout is not numeric") from error
    if not math.isfinite(timeout) or timeout <= 0:
        raise RuntimeError("owner-manifest verification timeout must be finite and positive")
    if path.is_symlink() or not path.is_file() or signature.is_symlink() or not signature.is_file():
        raise ValueError("ownership manifest or signature is unavailable or unsafe")
    raw = path.read_bytes()
    with tempfile.TemporaryDirectory(prefix="aidefend-owner-manifest-") as directory:
        snapshot = Path(directory) / "asset_owners.yaml"
        snapshot.write_bytes(raw)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", str(key),
                "--bundle", str(signature), str(snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout,
        )
    return raw


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--owners", type=Path, required=True)
    parser.add_argument("--owners-signature", type=Path, required=True)
    parser.add_argument("--owners-key", type=Path, required=True)
    parser.add_argument("--model-name", required=True)
    parser.add_argument("--model-version", required=True)
    parser.add_argument("--asset-digest", required=True)
    parser.add_argument("--apply", action="store_true")
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()

    if not DIGEST.fullmatch(args.asset_digest):
        raise ValueError("asset digest must be lowercase sha256:<64 hex>")
    manifest_bytes = verified_manifest_snapshot(
        args.owners, args.owners_signature, args.owners_key
    )
    try:
        manifest_text = manifest_bytes.decode("utf-8")
        document = yaml.load(manifest_text, Loader=UniqueKeySafeLoader)
    except (UnicodeError, yaml.YAMLError, ValueError) as error:
        raise ValueError("ownership manifest is not strict UTF-8 YAML") from error
    if not isinstance(document, dict) or document.get("schema_version") != "aidefend.asset-owners.v1":
        raise ValueError("ownership manifest schema differs")
    assets = document.get("assets")
    if not isinstance(assets, dict) or args.asset_digest not in assets:
        raise ValueError("exact asset digest has no owner record")
    owner = assets[args.asset_digest]
    if (
        not isinstance(owner, dict)
        or set(owner) != {"owner_id", "escalation", "status"}
        or not OWNER_ID.fullmatch(str(owner["owner_id"]))
        or not OWNER_ID.fullmatch(str(owner["escalation"]))
        or owner["status"] != "active"
    ):
        raise ValueError("owner record is incomplete or inactive")

    client = MlflowClient()
    model_version = client.get_model_version(args.model_name, args.model_version)
    downloaded = Path(mlflow.artifacts.download_artifacts(artifact_uri=model_version.source))
    observed_digest = tree_sha256(downloaded)
    if observed_digest != args.asset_digest:
        raise ValueError("downloaded registry artifact digest differs from ownership manifest")

    manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
    expected_tags = {
        "aidefend.owner_id": owner["owner_id"],
        "aidefend.owner_escalation": owner["escalation"],
        "aidefend.asset_digest": args.asset_digest,
        "aidefend.ownership_manifest_sha256": manifest_sha256,
    }
    if args.apply:
        for key, value in expected_tags.items():
            client.set_model_version_tag(args.model_name, args.model_version, key, value)

    readback = client.get_model_version(args.model_name, args.model_version)
    for key, expected in expected_tags.items():
        if readback.tags.get(key) != expected:
            raise PermissionError(f"model-version ownership tag differs: {key}")
    result = {
        "schema_version": "aidefend.asset-owner-binding.v1",
        "model_name": readback.name,
        "model_version": str(readback.version),
        "registry_source": readback.source,
        "asset_digest": observed_digest,
        "owner_id": owner["owner_id"],
        "owner_escalation": owner["escalation"],
        "ownership_manifest_sha256": manifest_sha256,
    }
    args.out.parent.mkdir(parents=True, exist_ok=True)
    args.out.write_text(json.dumps(result, sort_keys=True) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()

Step 3: Use separate writer and reader identities

set -euo pipefail
export M001001_OWNER_MANIFEST_VERIFY_TIMEOUT_SECONDS="${M001001_OWNER_MANIFEST_VERIFY_TIMEOUT_SECONDS:?injected from verified bootstrap profile}"
python tools/bind_model_owner.py --owners inventory/asset_owners.yaml --owners-signature inventory/asset_owners.yaml.sig --owners-key keys/asset-owner-authority.pub --model-name Fraud-Model --model-version 7 --asset-digest sha256:4f5e0d2a9d47d0e6c2df7c515d0e0c14b9f0d5f865f60f3c8bff2d99d5b3a481 --apply --out evidence/owner-binding-writer.json
# Re-run with a read-only MLflow credential after the writer credential is removed.
python tools/bind_model_owner.py --owners inventory/asset_owners.yaml --owners-signature inventory/asset_owners.yaml.sig --owners-key keys/asset-owner-authority.pub --model-name Fraud-Model --model-version 7 --asset-digest sha256:4f5e0d2a9d47d0e6c2df7c515d0e0c14b9f0d5f865f60f3c8bff2d99d5b3a481 --out evidence/owner-binding-reader.json
cmp evidence/owner-binding-writer.json evidence/owner-binding-reader.json
cosign sign-blob --yes --key env://OWNER_BINDING_VERIFIER_KEY --bundle evidence/owner-binding-reader.sig evidence/owner-binding-reader.json

Action: Reject an inventory release when the canonical digest has no active owner, the registry artifact bytes differ, the exact model version lacks the bound tags, or the read-only replay differs. Apply the same owner-ID contract to dataset and service registries through their native readback APIs.

AID-M-001.001-G004

Concept:

MLflow model stages are deprecated, but aliases and tags are also mutable operational metadata. Resolve the complete model population from a separately signed authoritative inventory/discovery snapshot, resolve each production alias, hash the actual model artifact bytes, and reconcile the exact set. Sign each canonical result outside the registry trust domain and durably accept it into append-only or retention-locked storage before updating the CMDB. Retention comes only from a verified signed policy. Neither environment-supplied name lists, MLflow, nor the CMDB is authoritative evidence.

Step 1: Resolve, measure, sign, and retain the complete governed population

from __future__ import annotations

# File: scripts/export_to_cmdb.py

import base64
import hashlib
import json
import math
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import quote, urlsplit

import boto3
import requests
from mlflow.artifacts import download_artifacts
from mlflow.tracking import MlflowClient


MLFLOW_TRACKING_URI = os.environ["MLFLOW_TRACKING_URI"]
CMDB_API_URL = os.environ["CMDB_API_URL"]
CMDB_API_TOKEN = os.environ["CMDB_API_TOKEN"]
PRODUCTION_ALIAS = "production"
EVIDENCE_BUCKET = os.environ["AI_INVENTORY_EVIDENCE_BUCKET"]
EVIDENCE_SIGNING_KEY_ID = os.environ["AI_INVENTORY_SIGNING_KMS_KEY_ID"]
EVIDENCE_STORAGE_KEY_ID = os.environ["AI_INVENTORY_STORAGE_KMS_KEY_ID"]
GOVERNED_POPULATION_PATH = Path(os.environ["AI_GOVERNED_POPULATION_PATH"])
GOVERNED_POPULATION_SIGNING_KEY_ID = os.environ["AI_GOVERNED_POPULATION_SIGNING_KMS_KEY_ID"]
INVENTORY_POLICY_PATH = Path(os.environ["AI_INVENTORY_SIGNED_POLICY_PATH"])
INVENTORY_POLICY_SIGNING_KEY_ID = os.environ["AI_INVENTORY_POLICY_SIGNING_KMS_KEY_ID"]
REQUIRED_TAGS = {"owner_team", "security_status", "training_dataset", "change_ticket"}

client = MlflowClient(tracking_uri=MLFLOW_TRACKING_URI)
kms = boto3.client("kms")
s3 = boto3.client("s3")


def canonical_bytes(value: dict[str, object]) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")


def reject_duplicate_keys(pairs):
    result = {}
    for key, item in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = item
    return result


def reject_nonfinite(token):
    raise ValueError(f"non-finite JSON number: {token}")


def validated_https_origin(raw: str, label: str) -> str:
    if raw != raw.strip():
        raise ValueError(f"{label} contains surrounding whitespace")
    try:
        parsed = urlsplit(raw)
        parsed.port
    except ValueError as error:
        raise ValueError(f"{label} is invalid") from error
    if (parsed.scheme.lower() != "https" or not parsed.hostname
            or parsed.username is not None or parsed.password is not None
            or parsed.path not in {"", "/"} or parsed.query or parsed.fragment):
        raise ValueError(f"{label} must be an exact HTTPS origin")
    return raw.rstrip("/")


CMDB_API_ORIGIN = validated_https_origin(CMDB_API_URL, "CMDB API URL")


def load_signed_document(path: Path, trusted_key_id: str, field: str) -> tuple[dict, str]:
    envelope = json.loads(
        path.read_text(encoding="utf-8"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=reject_nonfinite,
    )
    document = envelope[field]
    digest = hashlib.sha256(canonical_bytes(document)).digest()
    if envelope.get("payload_sha256") != digest.hex() or envelope.get("signing_key_id") != trusted_key_id:
        raise ValueError(f"{field} digest or signer mismatch")
    verified = kms.verify(
        KeyId=trusted_key_id,
        Message=digest,
        MessageType="DIGEST",
        Signature=base64.b64decode(envelope["signature"], validate=True),
        SigningAlgorithm="ECDSA_SHA_256",
    )
    if not verified.get("SignatureValid"):
        raise ValueError(f"{field} signature is invalid")
    return document, digest.hex()


def artifact_tree_sha256(source_uri: str) -> str:
    root = Path(download_artifacts(artifact_uri=source_uri))
    files = [root] if root.is_file() else sorted(path for path in root.rglob("*") if path.is_file())
    if not files:
        raise RuntimeError(f"model artifact is empty: {source_uri}")
    digest = hashlib.sha256()
    for path in files:
        relative = path.name if root.is_file() else path.relative_to(root).as_posix()
        name_bytes = relative.encode("utf-8")
        digest.update(len(name_bytes).to_bytes(8, "big"))
        digest.update(name_bytes)
        digest.update(path.stat().st_size.to_bytes(8, "big"))
        with path.open("rb") as handle:
            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                digest.update(chunk)
    return digest.hexdigest()


governed_population, governed_population_sha256 = load_signed_document(
    GOVERNED_POPULATION_PATH, GOVERNED_POPULATION_SIGNING_KEY_ID, "governed_population"
)
inventory_policy, inventory_policy_sha256 = load_signed_document(
    INVENTORY_POLICY_PATH, INVENTORY_POLICY_SIGNING_KEY_ID, "inventory_policy"
)
def positive_policy_integer(name: str) -> int:
    value = inventory_policy.get(name)
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"signed inventory-policy field must be a positive integer: {name}")
    return value


def positive_policy_number(name: str) -> float:
    value = inventory_policy.get(name)
    if (isinstance(value, bool) or not isinstance(value, (int, float))
            or not math.isfinite(float(value)) or float(value) <= 0):
        raise ValueError(f"signed inventory-policy field must be finite and positive: {name}")
    return float(value)


if not isinstance(inventory_policy.get("policy_version"), str) or not inventory_policy["policy_version"]:
    raise ValueError("signed inventory policy lacks a version")
RETENTION_DAYS = positive_policy_integer("evidence_retention_days")
CMDB_CONNECT_TIMEOUT_SECONDS = positive_policy_number("cmdb_connect_timeout_seconds")
CMDB_READ_TIMEOUT_SECONDS = positive_policy_number("cmdb_read_timeout_seconds")
population_entries = governed_population.get("models", [])
GOVERNED_MODELS = {entry.get("registry_name"): entry for entry in population_entries}
if not population_entries or None in GOVERNED_MODELS or len(GOVERNED_MODELS) != len(population_entries):
    raise ValueError("signed governed population is empty or has duplicate/missing model names")
for name, entry in GOVERNED_MODELS.items():
    expected_digest = str(entry.get("model_artifact_sha256", ""))
    if len(expected_digest) != 64 or any(character not in "0123456789abcdef" for character in expected_digest):
        raise ValueError(f"{name} lacks a valid expected model artifact digest")


def resolve_alias_state(
    model_entry: dict[str, object], *, measure_artifact: bool
) -> dict[str, object]:
    model_name = str(model_entry["registry_name"])
    version = client.get_model_version_by_alias(model_name, PRODUCTION_ALIAS)
    tags = dict(version.tags or {})
    missing = sorted(REQUIRED_TAGS - set(tags))
    if missing:
        raise RuntimeError(f"{model_name} v{version.version} missing required tags: {missing}")
    if tags["security_status"] != "approved":
        raise RuntimeError(f"{model_name} production alias does not reference an approved version")
    measured_artifact_digest = (
        artifact_tree_sha256(version.source)
        if measure_artifact
        else str(model_entry["model_artifact_sha256"])
    )
    if measured_artifact_digest != model_entry["model_artifact_sha256"]:
        raise RuntimeError(f"{model_name} artifact bytes differ from signed governed population")

    return {
        "schema_version": "aidefend.ai_inventory_snapshot.v1",
        "registry_name": model_name,
        "registry_version": str(version.version),
        "deployment_alias": PRODUCTION_ALIAS,
        "owner_team": tags["owner_team"],
        "security_status": tags["security_status"],
        "training_dataset": tags["training_dataset"],
        "change_ticket": tags["change_ticket"],
        "run_id": version.run_id,
        "source_uri": version.source,
        "model_artifact_sha256": measured_artifact_digest,
        "governed_population_sha256": governed_population_sha256,
        "inventory_policy_sha256": inventory_policy_sha256,
        "inventory_policy_version": inventory_policy["policy_version"],
        "last_updated_timestamp_ms": version.last_updated_timestamp,
    }


def retain_signed_snapshot(snapshot: dict[str, object]) -> dict[str, str]:
    payload = canonical_bytes(snapshot)
    payload_digest = hashlib.sha256(payload).digest()
    signed = kms.sign(
        KeyId=EVIDENCE_SIGNING_KEY_ID,
        Message=payload_digest,
        MessageType="DIGEST",
        SigningAlgorithm="ECDSA_SHA_256",
    )
    envelope = {
        "snapshot": snapshot,
        "snapshot_sha256": payload_digest.hex(),
        "signature": base64.b64encode(signed["Signature"]).decode("ascii"),
        "signing_key_id": EVIDENCE_SIGNING_KEY_ID,
        "signing_algorithm": "ECDSA_SHA_256",
    }
    body = canonical_bytes(envelope)
    body_checksum = base64.b64encode(hashlib.sha256(body).digest()).decode("ascii")
    retain_until = datetime.now(timezone.utc) + timedelta(days=RETENTION_DAYS)
    safe_name = quote(str(snapshot["registry_name"]), safe="")
    object_key = f"ai-inventory/{safe_name}/{payload_digest.hex()}.json"
    written = s3.put_object(
        Bucket=EVIDENCE_BUCKET,
        Key=object_key,
        Body=body,
        ContentType="application/json",
        ChecksumAlgorithm="SHA256",
        ChecksumSHA256=body_checksum,
        IfNoneMatch="*",
        ServerSideEncryption="aws:kms",
        SSEKMSKeyId=EVIDENCE_STORAGE_KEY_ID,
        ObjectLockMode="COMPLIANCE",
        ObjectLockRetainUntilDate=retain_until,
    )
    version_id = written.get("VersionId")
    if not version_id:
        raise RuntimeError("Evidence store did not return an immutable object version")
    accepted = s3.head_object(Bucket=EVIDENCE_BUCKET, Key=object_key, VersionId=version_id)
    if accepted.get("ObjectLockMode") != "COMPLIANCE":
        raise RuntimeError("Evidence object was not accepted under compliance retention")
    if accepted.get("ObjectLockRetainUntilDate") is None or accepted["ObjectLockRetainUntilDate"] < retain_until:
        raise RuntimeError("Evidence retention is absent or shorter than requested")
    return {
        "uri": f"s3://{EVIDENCE_BUCKET}/{object_key}?versionId={version_id}",
        "snapshot_sha256": payload_digest.hex(),
        "object_version": version_id,
        "signing_key_id": EVIDENCE_SIGNING_KEY_ID,
    }


def sync_models_to_cmdb() -> None:
    processed_models = set()
    with requests.Session() as session:
        session.headers.update(
            {
                "Authorization": f"Bearer {CMDB_API_TOKEN}",
                "Content-Type": "application/json",
            }
        )
        for model_name in sorted(GOVERNED_MODELS):
            model_entry = GOVERNED_MODELS[model_name]
            # Measure the artifact tree once. The cheap readbacks below re-fetch
            # alias metadata and tags without downloading the same model again.
            initial_state = resolve_alias_state(model_entry, measure_artifact=True)
            if resolve_alias_state(model_entry, measure_artifact=False) != initial_state:
                raise RuntimeError(f"{model_name} alias or tags drifted before evidence export")

            observed_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
            snapshot = {**initial_state, "observed_at": observed_at}
            evidence = retain_signed_snapshot(snapshot)

            if resolve_alias_state(model_entry, measure_artifact=False) != initial_state:
                raise RuntimeError(f"{model_name} alias or tags drifted during evidence export")

            ci_name = f"AI_MODEL_{model_name}_{initial_state['registry_version']}"
            payload = {
                "ci_name": ci_name,
                "category": "AI/ML Model",
                **initial_state,
                "authoritative_evidence": evidence,
            }
            with session.put(
                f"{CMDB_API_ORIGIN}/{quote(ci_name, safe='')}",
                json=payload,
                timeout=(CMDB_CONNECT_TIMEOUT_SECONDS, CMDB_READ_TIMEOUT_SECONDS),
                allow_redirects=False,
                stream=True,
            ) as response:
                response.raise_for_status()
            processed_models.add(model_name)
            print(f"Synchronized {ci_name} from {evidence['uri']}")

    if processed_models != set(GOVERNED_MODELS):
        raise RuntimeError("CMDB export did not reconcile the exact signed governed population")


if __name__ == "__main__":
    sync_models_to_cmdb()

The exporter identity must be separate from the MLflow service identity: MLflow may expose registry data but cannot choose the governed population, choose retention, use the evidence-signing key, shorten retention, overwrite evidence, or delete evidence versions. Equivalent external KMS/HSM plus append-only storage mechanisms are acceptable.

Step 2: Verify freshness and reconciliation independently

A read-only verifier recomputes the signed population and policy digests, the actual model artifact tree digest, and the canonical evidence digest; verifies configured signers and the exact retention-locked object version; and compares the current alias-resolved version, mandatory tags, artifact digest, exact population, and CMDB pointer. PASS requires every governed model and no omitted model to match. Drift or a partial/extra population is FAIL. Empty population, missing artifact bytes/digest, unavailable signed policy/population, or unverifiable evidence is INSUFFICIENT_DATA and fails the inventory/release gate.

Action: Trigger the export after every authoritative population, release artifact, or alias change and on scheduled reconciliation. Never count an environment name list, MLflow alias/tag, source URI, CMDB row, exporter exit code, or tool presence alone as durable or complete evidence.

Open-source tools

• MLflow
• Kubeflow (in-platform pipeline, model, artifact, and deployment-status metadata inventory; authoritative ownership and observed-state reconciliation remain external)
• DVC
• Great Expectations
• AWS CLI v2
• Azure CLI
• Snipe-IT

Commercial / hosted tools

• Wiz AI-SPM
• Microsoft Defender for Cloud
• Palo Alto Networks Prisma Cloud
• Amazon SageMaker AI Model Registry
• Google Vertex AI Model Registry
• Databricks Unity Catalog
• Alation Data Intelligence Platform
• Collibra Data Intelligence Platform
• Google Cloud CLI

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• N/A

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AITech-6.1 Training Data Poisoning (Signed dataset digest is checked before the illustrated training run)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Governance 4.1: Lack of traceability and transparency of model assets
• Governance 4.2: Lack of end-to-end ML lifecycle
• Model Management 8.1: Model attribution
• Operations 11.1: Lack of MLOps - repeatable enforced standards