Scope boundary

Owns signed, version-bound reference explanations for a governed benign input population, including explainer configuration, feature attributions, stability expectations, model digest, and population digest.

AID-D-006: AID-M-003.003 creates and recalibrates the trusted XAI output reference; AID-D-006 performs runtime comparison and triage of explanation drift. An explanation deviation is a Detect-side investigative signal, while the baseline control does not decide that an attack occurred.

Implementation guidance

• AID-M-003.003-G001: Generate and store baseline feature attributions for different prediction classes.
• AID-M-003.003-G002: Create qualitative documentation of expected explanatory behavior in model cards.
• AID-M-003.003-G003: Baseline the stability of explanations under minor input perturbations.
• AID-M-003.003-G004: Version control XAI baselines and link them to specific model versions in a registry.

AID-M-003.003-G001

Concept:

We compute SHAP values on a trusted reference dataset and average the absolute attribution per feature. That gives us a reproducible 'this is what matters' fingerprint for the model. We store it as JSON so we can diff later. Below is a runnable-style script: it imports needed libs, assumes a scikit-learn style model, and uses a DataFrame so .columns is defined.

from __future__ import annotations

# File: modeling/generate_xai_baselines.py

import hashlib
import json
import os
import platform
import io
import tempfile
from pathlib import Path

import numpy as np
from skops.io import load as load_skops
import pandas as pd
import shap

MODEL_PATH = Path("artifacts/model.skops")
BASELINE_DATA_PATH = Path("data/X_baseline.csv")
OUTPUT_PATH = Path("baselines/model_v2_xai_baseline.json")
CLASS_INDEX = 1
RELEASE_ID = os.environ["VERIFIED_RELEASE_ID"].strip()
if not RELEASE_ID:
    raise ValueError("verified release identity is required")


def sha256_file(path: Path) -> str:
    return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest()


def capture_baseline_inputs(model_path: Path, baseline_path: Path):
    # Load and hash the same captured bytes even if an input path later changes.
    model_raw = model_path.read_bytes()
    baseline_raw = baseline_path.read_bytes()
    if not model_raw or not baseline_raw:
        raise ValueError("model and baseline bytes must be nonempty")
    with tempfile.TemporaryDirectory(prefix="xai-source-") as directory:
        snapshot = Path(directory) / "model.skops"
        descriptor = os.open(snapshot, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400)
        with os.fdopen(descriptor, "wb") as handle:
            handle.write(model_raw)
            handle.flush()
            os.fsync(handle.fileno())
        model = load_skops(snapshot, trusted=[])
    dataset = pd.read_csv(io.BytesIO(baseline_raw))
    return model, dataset, model_raw, baseline_raw


def select_class_attributions(explanation, *, class_index: int, feature_count: int) -> np.ndarray:
    values = explanation.values
    if isinstance(values, list):
        if class_index >= len(values):
            raise ValueError("class_index is outside SHAP output list")
        selected = np.asarray(values[class_index], dtype=float)
    else:
        arr = np.asarray(values, dtype=float)
        if arr.ndim == 2 and arr.shape[1] == feature_count:
            selected = arr
        elif arr.ndim == 3 and arr.shape[1] == feature_count:
            selected = arr[:, :, class_index]
        elif arr.ndim == 3 and arr.shape[2] == feature_count:
            selected = arr[:, class_index, :]
        else:
            raise ValueError(f"unsupported SHAP values shape {arr.shape}")
    if selected.ndim != 2 or selected.shape[1] != feature_count:
        raise ValueError(f"selected attribution shape {selected.shape} does not match feature count {feature_count}")
    if not np.isfinite(selected).all():
        raise ValueError("SHAP attributions contain NaN or infinity")
    return selected


model, X_baseline, model_raw, baseline_raw = capture_baseline_inputs(MODEL_PATH, BASELINE_DATA_PATH)
if X_baseline.empty:
    raise ValueError("baseline data is empty")
feature_names = list(X_baseline.columns)
feature_schema = [
    {"name": str(name), "dtype": str(X_baseline[name].dtype)}
    for name in feature_names
]

explainer = shap.Explainer(model.predict_proba, X_baseline)
shap_values = explainer(X_baseline)
values_for_class = np.abs(
    select_class_attributions(
        shap_values, class_index=CLASS_INDEX, feature_count=len(feature_names)
    )
)
avg_feature_importance = values_for_class.mean(axis=0)

xai_baseline = {
    "schema_version": "aidefend.xai-baseline.v1",
    "release_id": RELEASE_ID,
    "method": "SHAP",
    "explainer_class": type(explainer).__name__,
    "class_of_interest": CLASS_INDEX,
    "model_sha256": "sha256:" + hashlib.sha256(model_raw).hexdigest(),
    "baseline_dataset_sha256": "sha256:" + hashlib.sha256(baseline_raw).hexdigest(),
    "baseline_rows": int(len(X_baseline)),
    "feature_schema": feature_schema,
    "feature_schema_sha256": "sha256:" + hashlib.sha256(
        json.dumps(feature_schema, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest(),
    "feature_count": len(feature_names),
    "runtime": {
        "python": platform.python_version(),
        "numpy": np.__version__,
        "pandas": pd.__version__,
        "shap": shap.__version__,
    },
    "average_feature_importance": {
        fname: float(score)
        for fname, score in zip(feature_names, avg_feature_importance, strict=True)
    },
}

OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_text(
    json.dumps(xai_baseline, indent=2, sort_keys=True), encoding="utf-8"
)
print(f"XAI baseline saved to {OUTPUT_PATH}")

Action: Run this script during validation. It fails closed on unsupported SHAP tensor layouts instead of silently treating a feature axis as a class axis. The local JSON is only a digest-bound candidate baseline; it cannot produce control PASS. A governed baseline registry must authenticate the publisher, bind the exact candidate bytes to the release, apply policy-owned currentness, publish an immutable version, and return a readback of the same model, dataset, feature schema, explainer/runtime, and baseline digests. Missing, stale, revoked, mismatched, or unreadback publication remains INSUFFICIENT_DATA or ERROR.

AID-M-003.003-G002

Concept:

Numbers alone don't capture business intuition. We also add a human-readable 'Expected Explanations' section to the model card that states which features should logically drive a 'fraud' vs 'not fraud' decision. This is crucial for abuse investigations and regulatory review.

# In your model_card.md or in the generated model card's free-text section:

## Expected Explanatory Behavior (SHAP)

- For 'Fraud' predictions:
  We expect features like `transaction_amount`, `hours_since_last_login`,
  and `num_failed_logins_24h` to have high positive SHAP values.

- For 'Not Fraud' predictions:
  We expect features like `user_has_mfa_enabled` and `is_known_device`
  to have strong negative SHAP values (they push score away from fraud).

- Red flag:
  If we see irrelevant features (e.g. `user_id_hash`, raw timestamp entropy)
  showing up as top drivers, that may indicate data poisoning or prompt abuse
  and must be escalated.

Action: Treat this as required governance text. It lets fraud ops / compliance quickly tell if the model is reasoning in a sane way or has been steered into weird features.

Before you begin

Apply this method when the signed model assurance policy requires explainability evidence for the model type and decision context.

AID-M-003.003-G003

Concept

Explanation-stability evidence must be reproducible. Resolve perturbation scales, sample size, seeds, prediction tolerance, and acceptance floors from a signed model-assurance policy; bind the result to the exact model, data, policy, and implementation bytes. This runnable variant applies to numeric tabular predictors.

Verify policy inputs and compute every declared case

The signed policy has schema aidefend.xai-stability-policy.v1 and contains exact model/data paths and SHA-256 digests, ordered feature names, a per-feature noise standard deviation, sample count and seed, a non-empty repetition-seed population, prediction-delta tolerance, and mean/p10 stability floors. These values are environment policy, not framework defaults.

# File: modeling/calculate_xai_stability.py
from __future__ import annotations

import hashlib
import io
import json
import math
import os
import re
import stat
from pathlib import Path

import numpy as np
import pandas as pd
import shap
from scipy.stats import spearmanr
from skops.io import loads as load_skops

SHA256 = re.compile(r"^[0-9a-f]{64}$")


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


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


def regular_bytes(path: Path, maximum: int | None = None) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode) or info.st_size < 1:
            raise ValueError(f"input is empty or not a regular file: {path}")
        if maximum is not None and info.st_size > maximum:
            raise ValueError(f"input exceeds signed byte bound: {path}")
        with os.fdopen(descriptor, "rb", closefd=False) as handle:
            raw = handle.read(info.st_size + 1)
        if len(raw) != info.st_size:
            raise ValueError(f"input changed during read: {path}")
        return raw
    finally:
        os.close(descriptor)


def sha256(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


policy_path = Path(os.environ["XAI_STABILITY_POLICY"])
policy_raw = regular_bytes(policy_path)
policy = json.loads(
    policy_raw.decode("utf-8", errors="strict"),
    object_pairs_hook=unique_object,
    parse_constant=reject_nonfinite,
)
required = {
    "schema_version", "policy_version", "model_path", "model_sha256",
    "baseline_data_path", "baseline_data_sha256", "feature_columns",
    "feature_noise_std", "sample_count", "sample_seed",
    "repetition_seeds", "max_prediction_delta",
    "mean_stability_floor", "p10_stability_floor",
    "trusted_model_types", "max_model_bytes", "max_data_bytes",
}
if not isinstance(policy, dict) or set(policy) != required:
    raise ValueError("XAI stability policy fields differ")
if policy["schema_version"] != "aidefend.xai-stability-policy.v1":
    raise ValueError("XAI stability policy schema differs")
if (
    not isinstance(policy["policy_version"], str)
    or not policy["policy_version"]
    or not isinstance(policy["model_path"], str)
    or not policy["model_path"]
    or not isinstance(policy["baseline_data_path"], str)
    or not policy["baseline_data_path"]
    or not SHA256.fullmatch(policy["model_sha256"])
    or not SHA256.fullmatch(policy["baseline_data_sha256"])
):
    raise ValueError("XAI policy identity or digest is invalid")
if (
    not isinstance(policy["feature_columns"], list)
    or not policy["feature_columns"]
    or any(not isinstance(name, str) or not name for name in policy["feature_columns"])
    or len(policy["feature_columns"]) != len(set(policy["feature_columns"]))
):
    raise ValueError("feature population is empty, untyped, or duplicated")
if (
    not isinstance(policy["feature_noise_std"], dict)
    or set(policy["feature_noise_std"]) != set(policy["feature_columns"])
):
    raise ValueError("per-feature noise population differs")
if (
    not isinstance(policy["trusted_model_types"], list)
    or any(not isinstance(name, str) or not name for name in policy["trusted_model_types"])
    or len(policy["trusted_model_types"]) != len(set(policy["trusted_model_types"]))
):
    raise ValueError("trusted model-type allow-list is invalid")
for field in ("max_model_bytes", "max_data_bytes"):
    bound = policy[field]
    if isinstance(bound, bool) or not isinstance(bound, int) or bound < 1:
        raise ValueError(f"signed byte bound is invalid: {field}")
if (
    isinstance(policy["sample_count"], bool)
    or not isinstance(policy["sample_count"], int)
    or policy["sample_count"] < 1
    or isinstance(policy["sample_seed"], bool)
    or not isinstance(policy["sample_seed"], int)
    or not isinstance(policy["repetition_seeds"], list)
    or not policy["repetition_seeds"]
    or any(isinstance(seed, bool) or not isinstance(seed, int) for seed in policy["repetition_seeds"])
    or len(policy["repetition_seeds"]) != len(set(policy["repetition_seeds"]))
):
    raise ValueError("sample or repetition population is invalid")
for field in ("max_prediction_delta", "mean_stability_floor", "p10_stability_floor"):
    value = policy[field]
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
        raise ValueError(f"policy value is invalid: {field}")
if policy["max_prediction_delta"] < 0:
    raise ValueError("prediction tolerance cannot be negative")
for field in ("mean_stability_floor", "p10_stability_floor"):
    if policy[field] < -1 or policy[field] > 1:
        raise ValueError(f"rank-correlation floor is outside [-1, 1]: {field}")
for feature, value in policy["feature_noise_std"].items():
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:
        raise ValueError(f"noise scale is invalid: {feature}")

model_path = Path(policy["model_path"])
data_path = Path(policy["baseline_data_path"])
model_raw = regular_bytes(model_path, policy["max_model_bytes"])
data_raw = regular_bytes(data_path, policy["max_data_bytes"])
if sha256(model_raw) != policy["model_sha256"] or sha256(data_raw) != policy["baseline_data_sha256"]:
    raise ValueError("model or reference-data digest differs")
model = load_skops(model_raw, trusted=policy["trusted_model_types"])
frame = pd.read_csv(io.BytesIO(data_raw))
if list(frame.columns) != policy["feature_columns"]:
    raise ValueError("reference feature order differs")
if len(frame) < policy["sample_count"]:
    raise ValueError("reference population is smaller than signed sample count")
sampled = frame.sample(n=policy["sample_count"], random_state=policy["sample_seed"])
matrix = sampled.to_numpy(dtype=float)
if not np.isfinite(matrix).all():
    raise ValueError("reference population contains non-finite values")

explainer = shap.Explainer(model.predict, sampled)
noise_scale = np.asarray(
    [policy["feature_noise_std"][name] for name in policy["feature_columns"]],
    dtype=float,
)
cases = []
for seed in policy["repetition_seeds"]:
    rng = np.random.default_rng(seed)
    for row_number, row in enumerate(matrix):
        perturbed = row + rng.normal(0.0, noise_scale, size=row.shape)
        base_prediction = np.asarray(model.predict(row.reshape(1, -1)), dtype=float).reshape(-1)
        perturbed_prediction = np.asarray(model.predict(perturbed.reshape(1, -1)), dtype=float).reshape(-1)
        prediction_delta = float(np.max(np.abs(base_prediction - perturbed_prediction)))
        base_values = np.asarray(explainer(row.reshape(1, -1)).values, dtype=float).reshape(-1)
        perturbed_values = np.asarray(explainer(perturbed.reshape(1, -1)).values, dtype=float).reshape(-1)
        correlation, _ = spearmanr(np.abs(base_values), np.abs(perturbed_values))
        stability = 0.0 if not np.isfinite(correlation) else float(correlation)
        cases.append({
            "seed": seed,
            "sample_row": row_number,
            "prediction_delta": prediction_delta,
            "prediction_stable": prediction_delta <= policy["max_prediction_delta"],
            "spearman_rank_stability": stability,
        })

expected_count = policy["sample_count"] * len(policy["repetition_seeds"])
if len(cases) != expected_count:
    raise RuntimeError("measured XAI population differs")
scores = np.asarray([item["spearman_rank_stability"] for item in cases], dtype=float)
unstable = sum(not item["prediction_stable"] for item in cases)
report = {
    "schema_version": "aidefend.xai-stability-result.v1",
    "policy_version": policy["policy_version"],
    "policy_sha256": sha256(policy_raw),
    "model_sha256": policy["model_sha256"],
    "baseline_data_sha256": policy["baseline_data_sha256"],
    "implementation_sha256": sha256(regular_bytes(Path(__file__))),
    "expected_case_count": expected_count,
    "measured_case_count": len(cases),
    "unstable_prediction_count": unstable,
    "mean_spearman_rank_stability": float(np.mean(scores)),
    "p10_spearman_rank_stability": float(np.percentile(scores, 10)),
    "cases": cases,
}
passed = (
    unstable == 0
    and report["mean_spearman_rank_stability"] >= policy["mean_stability_floor"]
    and report["p10_spearman_rank_stability"] >= policy["p10_stability_floor"]
)
report["outcome"] = "PASS" if passed else "FAIL"
output = Path("baselines/model_xai_stability.json")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
readback = json.loads(
    output.read_text(encoding="utf-8"),
    object_pairs_hook=unique_object,
    parse_constant=reject_nonfinite,
)
if readback != report:
    raise RuntimeError("XAI stability report readback differs")
if not passed:
    raise SystemExit("XAI stability population failed the signed policy")

Run from the verified policy snapshot

set -euo pipefail
umask 077
snapshot_dir="$(mktemp -d)"
trap 'rm -rf -- "$snapshot_dir"' EXIT
install -m 0400 -- "$XAI_STABILITY_POLICY" "$snapshot_dir/policy.json"
install -m 0400 -- "$XAI_STABILITY_POLICY_BUNDLE" "$snapshot_dir/policy.sigstore.json"
cosign verify-blob --key "$XAI_STABILITY_VERIFY_KEY"   --bundle "$snapshot_dir/policy.sigstore.json" "$snapshot_dir/policy.json" >/dev/null
export XAI_STABILITY_POLICY="$snapshot_dir/policy.json"
python modeling/calculate_xai_stability.py
cosign sign-blob --yes --key "$XAI_RESULT_SIGNING_KEY"   --bundle baselines/model_xai_stability.sigstore.json   baselines/model_xai_stability.json

Action: Re-run the same signed case population to reproduce the report before promotion. A changed seed, silently reduced sample set, unstable prediction, non-finite score, missing case, or policy/model/data digest mismatch fails the baseline rather than being averaged away.

AID-M-003.003-G004

Concept and prerequisite

Bind the signed XAI baseline bytes to the exact MLflow registered-model name and version, not merely to a neighboring run. Pin the MLflow Python client and tracking-server compatibility tuple in the deployment lock, use an authenticated tracking endpoint, and supply all paths and timeouts from the signed release profile.

Step 1: Register the model and publish a signed version binding

# File: modeling/register_xai_bundle.py
from __future__ import annotations

import hashlib
import json
import math
import os
import stat
import subprocess
import tempfile
from pathlib import Path

import mlflow
import mlflow.artifacts
import mlflow.sklearn
from mlflow import MlflowClient
from skops.io import load as load_skops


def positive_timeout(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be finite and positive")
    return value


COMMAND_TIMEOUT = positive_timeout("M003003_COMMAND_TIMEOUT_SECONDS")
REGISTRATION_TIMEOUT = int(os.environ["M003003_REGISTRATION_TIMEOUT_SECONDS"])
if REGISTRATION_TIMEOUT < 1:
    raise RuntimeError("registration timeout must be positive")
MODEL_PATH = Path(os.environ["VERIFIED_MODEL_ARTIFACT_PATH"])
BASELINE_PATH = Path(os.environ["VERIFIED_XAI_BASELINE_PATH"])
BASELINE_SIGNATURE = Path(os.environ["VERIFIED_XAI_BASELINE_SIGNATURE"])
MODEL_NAME = os.environ["MLFLOW_REGISTERED_MODEL_NAME"]
if not MODEL_NAME.strip():
    raise RuntimeError("registered model name is required")


def sha256_file(path: Path) -> str:
    if not path.is_file() or path.is_symlink() or path.stat().st_size == 0:
        raise ValueError(f"missing, empty, or unsafe file: {path}")
    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 not path.exists() or path.is_symlink():
        raise ValueError(f"missing or unsafe artifact: {path}")
    files = [path] if path.is_file() else sorted(
        item for item in path.rglob("*") if item.is_file()
    )
    if not files:
        raise ValueError(f"artifact contains no files: {path}")
    digest = hashlib.sha256()
    for item in files:
        if item.is_symlink():
            raise ValueError(f"artifact contains a symlink: {item}")
        relative = item.name if path.is_file() else item.relative_to(path).as_posix()
        name = relative.encode("utf-8")
        content = item.read_bytes()
        digest.update(len(name).to_bytes(8, "big"))
        digest.update(name)
        digest.update(len(content).to_bytes(8, "big"))
        digest.update(content)
    return digest.hexdigest()


def verify_blob(path: Path, signature: Path, key: str) -> None:
    subprocess.run(
        ["cosign", "verify-blob", "--key", key, "--bundle", str(signature), str(path)],
        check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT,
    )


def regular_bytes(path: Path) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    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 ValueError(f"missing, empty, or unsafe signed input: {path}")
        return handle.read()


def private_file(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 strict_json(raw: bytes, label: str) -> object:
    def unique(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
    return json.loads(
        raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)),
    )


snapshot_directory = tempfile.TemporaryDirectory(prefix="aidefend-xai-baseline-")
snapshot_root = Path(snapshot_directory.name)
snapshot_root.chmod(0o700)
baseline_raw = regular_bytes(BASELINE_PATH)
baseline_signature_raw = regular_bytes(BASELINE_SIGNATURE)
baseline_snapshot = snapshot_root / BASELINE_PATH.name
baseline_signature_snapshot = snapshot_root / BASELINE_SIGNATURE.name
if baseline_snapshot == baseline_signature_snapshot:
    raise ValueError("baseline payload and signature names collide")
private_file(baseline_snapshot, baseline_raw)
private_file(baseline_signature_snapshot, baseline_signature_raw)
verify_blob(baseline_snapshot, baseline_signature_snapshot, os.environ["XAI_BASELINE_VERIFY_KEY"])
baseline = strict_json(baseline_raw, "signed XAI baseline")
if not isinstance(baseline, dict) or set(baseline) != {
    "schema_version", "release_id", "method", "explainer_class", "class_of_interest",
    "model_sha256", "baseline_dataset_sha256", "baseline_rows", "feature_schema",
    "feature_schema_sha256", "feature_count", "runtime", "average_feature_importance",
}:
    raise ValueError("XAI baseline schema differs")
if (
    baseline["schema_version"] != "aidefend.xai-baseline.v1"
    or not isinstance(baseline["release_id"], str) or not baseline["release_id"]
    or baseline["release_id"] != os.environ["VERIFIED_RELEASE_ID"].strip()
    or not isinstance(baseline["model_sha256"], str)
    or not baseline["model_sha256"].startswith("sha256:")
    or len(baseline["model_sha256"]) != 71
    or set(baseline["model_sha256"][7:]) - set("0123456789abcdef")
):
    raise ValueError("XAI baseline identity is invalid")

# G001 hashes the exact .skops file, not a tree hash or MLflow directory.
source_model_raw = regular_bytes(MODEL_PATH)
source_model_sha256 = hashlib.sha256(source_model_raw).hexdigest()
model_snapshot = snapshot_root / "source-model.skops"
private_file(model_snapshot, source_model_raw)
if baseline["model_sha256"] != "sha256:" + source_model_sha256:
    raise ValueError("XAI baseline is bound to different source-model bytes")

mlflow.set_tracking_uri(os.environ["MLFLOW_TRACKING_URI"])
mlflow.set_experiment(os.environ["MLFLOW_EXPERIMENT_NAME"])
model = load_skops(model_snapshot, trusted=[])
client = MlflowClient()
binding_path = Path("artifacts/xai-binding.json")
binding_signature = binding_path.with_suffix(".sig")
binding_path.parent.mkdir(parents=True, exist_ok=True)

with mlflow.start_run() as run:
    mlflow.sklearn.log_model(model, name="classifier")
    mlflow.log_artifact(str(baseline_snapshot), artifact_path="xai")
    mlflow.log_artifact(str(baseline_signature_snapshot), artifact_path="xai")
    registered = mlflow.register_model(
        model_uri=f"runs:/{run.info.run_id}/classifier",
        name=MODEL_NAME,
        await_registration_for=REGISTRATION_TIMEOUT,
    )
    registered_uri = f"models:/{registered.name}/{registered.version}"
    registered_model_path = Path(
        mlflow.artifacts.download_artifacts(artifact_uri=registered_uri)
    )
    registered_model_sha256 = tree_sha256(registered_model_path)
    binding = {
        "schema_version": "aidefend.mlflow-xai-binding.v1",
        "release_id": baseline["release_id"],
        "model_name": registered.name,
        "model_version": str(registered.version),
        "model_run_id": run.info.run_id,
        "model_source_uri": registered.source,
        "model_source_sha256": source_model_sha256,
        "registered_model_sha256": registered_model_sha256,
        "baseline_artifact_path": "xai/" + baseline_snapshot.name,
        "baseline_signature_path": "xai/" + baseline_signature_snapshot.name,
        "baseline_sha256": hashlib.sha256(baseline_raw).hexdigest(),
        "baseline_signature_sha256": hashlib.sha256(baseline_signature_raw).hexdigest(),
    }
    binding_path.write_text(
        json.dumps(binding, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n",
        encoding="utf-8",
    )
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://XAI_BINDING_SIGNING_KEY",
            "--bundle", str(binding_signature), str(binding_path),
        ],
        check=True, timeout=COMMAND_TIMEOUT,
    )
    verify_blob(binding_path, binding_signature, os.environ["XAI_BINDING_VERIFY_KEY"])
    mlflow.log_artifact(str(binding_path), artifact_path="xai")
    mlflow.log_artifact(str(binding_signature), artifact_path="xai")

binding_sha256 = sha256_file(binding_path)
expected_tags = {
    "aidefend.release_id": binding["release_id"],
    "aidefend.xai_binding_path": "xai/" + binding_path.name,
    "aidefend.xai_binding_signature_path": "xai/" + binding_signature.name,
    "aidefend.xai_binding_sha256": binding_sha256,
    "aidefend.xai_baseline_sha256": binding["baseline_sha256"],
    "aidefend.registered_model_sha256": binding["registered_model_sha256"],
}
for key, value in expected_tags.items():
    client.set_model_version_tag(registered.name, registered.version, key, value)
readback = client.get_model_version(registered.name, registered.version)
if str(readback.version) != str(registered.version) or readback.run_id != binding["model_run_id"]:
    raise RuntimeError("registered-model version readback differs")
for key, expected in expected_tags.items():
    if readback.tags.get(key) != expected:
        raise RuntimeError(f"registered-model tag readback differs: {key}")
Path("artifacts/xai-registration.json").write_text(
    json.dumps({
        "model_name": readback.name,
        "model_version": str(readback.version),
        "binding_sha256": binding_sha256,
    }, sort_keys=True) + "\n",
    encoding="utf-8",
)

Step 2: Re-download through the exact registered-model version

# File: modeling/verify_xai_binding.py
from __future__ import annotations

import hashlib
import json
import math
import os
import stat
import subprocess
import tempfile
from pathlib import Path

import mlflow
import mlflow.artifacts
from mlflow import MlflowClient


def positive_timeout(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be finite and positive")
    return value


COMMAND_TIMEOUT = positive_timeout("M003003_COMMAND_TIMEOUT_SECONDS")


def sha256_file(path: Path) -> str:
    if not path.is_file() or path.is_symlink() or path.stat().st_size == 0:
        raise ValueError(f"missing, empty, or unsafe file: {path}")
    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 not path.exists() or path.is_symlink():
        raise ValueError(f"missing or unsafe artifact: {path}")
    files = [path] if path.is_file() else sorted(
        item for item in path.rglob("*") if item.is_file()
    )
    if not files:
        raise ValueError(f"artifact contains no files: {path}")
    digest = hashlib.sha256()
    for item in files:
        if item.is_symlink():
            raise ValueError(f"artifact contains a symlink: {item}")
        relative = item.name if path.is_file() else item.relative_to(path).as_posix()
        name = relative.encode("utf-8")
        content = item.read_bytes()
        digest.update(len(name).to_bytes(8, "big"))
        digest.update(name)
        digest.update(len(content).to_bytes(8, "big"))
        digest.update(content)
    return digest.hexdigest()


def verify_blob(path: Path, signature: Path, key: str) -> None:
    subprocess.run(
        ["cosign", "verify-blob", "--key", key, "--bundle", str(signature), str(path)],
        check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT,
    )


def regular_bytes(path: Path) -> bytes:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    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 ValueError(f"missing, empty, or unsafe signed input: {path}")
        return handle.read()


def private_file(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(path: Path, signature: Path, key: str) -> tuple[bytes, bytes]:
    raw, signature_raw = regular_bytes(path), regular_bytes(signature)
    with tempfile.TemporaryDirectory(prefix="aidefend-xai-readback-") as directory:
        root = Path(directory)
        root.chmod(0o700)
        payload_snapshot, signature_snapshot = root / "payload", root / "payload.sig"
        private_file(payload_snapshot, raw)
        private_file(signature_snapshot, signature_raw)
        verify_blob(payload_snapshot, signature_snapshot, key)
        if regular_bytes(payload_snapshot) != raw or regular_bytes(signature_snapshot) != signature_raw:
            raise ValueError("private signed-input snapshot changed during verification")
    return raw, signature_raw


def strict_json(raw: bytes, label: str) -> object:
    def unique(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
    return json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
                      parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)))

model_name = os.environ["MLFLOW_REGISTERED_MODEL_NAME"]
model_version = os.environ["MLFLOW_REGISTERED_MODEL_VERSION"]
mlflow.set_tracking_uri(os.environ["MLFLOW_TRACKING_URI"])
client = MlflowClient()
version = client.get_model_version(model_name, model_version)
if version.name != model_name or str(version.version) != model_version or not version.run_id:
    raise RuntimeError("exact registered-model version is unavailable")

required_tags = {
    "aidefend.release_id", "aidefend.xai_binding_path",
    "aidefend.xai_binding_signature_path", "aidefend.xai_binding_sha256",
    "aidefend.xai_baseline_sha256", "aidefend.registered_model_sha256",
}
if any(not version.tags.get(tag) for tag in required_tags):
    raise RuntimeError("exact registered-model version lacks XAI binding tags")

binding_path = Path(mlflow.artifacts.download_artifacts(
    run_id=version.run_id,
    artifact_path=version.tags["aidefend.xai_binding_path"],
))
binding_signature = Path(mlflow.artifacts.download_artifacts(
    run_id=version.run_id,
    artifact_path=version.tags["aidefend.xai_binding_signature_path"],
))
binding_raw, binding_signature_raw = capture_verified(
    binding_path, binding_signature, os.environ["XAI_BINDING_VERIFY_KEY"]
)
if hashlib.sha256(binding_raw).hexdigest() != version.tags["aidefend.xai_binding_sha256"]:
    raise RuntimeError("downloaded XAI binding digest differs from model-version tag")
binding = strict_json(binding_raw, "signed XAI binding")
required_binding = {
    "schema_version", "release_id", "model_name", "model_version",
    "model_run_id", "model_source_uri", "model_source_sha256",
    "registered_model_sha256", "baseline_artifact_path",
    "baseline_signature_path", "baseline_sha256",
    "baseline_signature_sha256",
}
if set(binding) != required_binding or binding["schema_version"] != "aidefend.mlflow-xai-binding.v1":
    raise RuntimeError("XAI binding schema differs")
if (
    binding["model_name"] != version.name
    or binding["model_version"] != str(version.version)
    or binding["model_run_id"] != version.run_id
    or binding["model_source_uri"] != version.source
    or binding["release_id"] != version.tags["aidefend.release_id"]
):
    raise RuntimeError("XAI binding does not name the exact registered-model version")

baseline_path = Path(mlflow.artifacts.download_artifacts(
    run_id=version.run_id, artifact_path=binding["baseline_artifact_path"]
))
baseline_signature = Path(mlflow.artifacts.download_artifacts(
    run_id=version.run_id, artifact_path=binding["baseline_signature_path"]
))
baseline_raw, baseline_signature_raw = capture_verified(
    baseline_path, baseline_signature, os.environ["XAI_BASELINE_VERIFY_KEY"]
)
if (
    hashlib.sha256(baseline_raw).hexdigest() != binding["baseline_sha256"]
    or hashlib.sha256(baseline_signature_raw).hexdigest() != binding["baseline_signature_sha256"]
    or binding["baseline_sha256"] != version.tags["aidefend.xai_baseline_sha256"]
):
    raise RuntimeError("XAI baseline bytes or signature differ from the exact version binding")
baseline = strict_json(baseline_raw, "signed XAI baseline")
if (
    baseline.get("schema_version") != "aidefend.xai-baseline.v1"
    or baseline.get("release_id") != binding["release_id"]
    or baseline.get("model_sha256") != "sha256:" + binding["model_source_sha256"]
):
    raise RuntimeError("XAI baseline identity differs from signed version binding")

registered_model = Path(mlflow.artifacts.download_artifacts(
    artifact_uri=f"models:/{version.name}/{version.version}"
))
observed_model_sha256 = tree_sha256(registered_model)
if (
    observed_model_sha256 != binding["registered_model_sha256"]
    or observed_model_sha256 != version.tags["aidefend.registered_model_sha256"]
):
    raise RuntimeError("registered-model bytes differ from XAI binding")
result = {
    "schema_version": "aidefend.mlflow-xai-readback.v1",
    "model_name": version.name,
    "model_version": str(version.version),
    "model_run_id": version.run_id,
    "release_id": binding["release_id"],
    "registered_model_sha256": observed_model_sha256,
    "baseline_sha256": binding["baseline_sha256"],
    "binding_sha256": hashlib.sha256(binding_raw).hexdigest(),
    "outcome": "PASS",
}
out = Path("evidence/mlflow-xai-readback.json")
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(result, sort_keys=True) + "\n", encoding="utf-8")
subprocess.run(
    [
        "cosign", "sign-blob", "--yes", "--key", "env://XAI_READBACK_SIGNING_KEY",
        "--bundle", str(out.with_suffix(".sig")), str(out),
    ],
    check=True, timeout=COMMAND_TIMEOUT,
)

Step 3: Run with separated registry roles

The registration job may create a model version and set its tags. A separate read-only MLflow identity runs the second script using the name and version from the signed release manifest, downloads the model through models:/name/version, re-downloads the baseline and signed binding from that version's run, and signs the readback. Missing bytes, a tag-only assertion, a different run or version, a failed signature, or a digest mismatch prevents release.

Action: Store the signed baseline, binding, and readback with the exact release. Do not infer a baseline-to-model relationship from run adjacency, mutable aliases, or a successful registration message.

G001 and this registration path share the detailed aidefend.xai-baseline.v1 SHAP schema. VERIFIED_MODEL_ARTIFACT_PATH names the same .skops bytes measured by G001; it is not an MLflow model directory. The signed release supplies VERIFIED_RELEASE_ID. The original source file uses a raw SHA-256 (with sha256: in the baseline); only the newly registered MLflow directory uses tree_sha256. Neither artifact may be substituted merely because predictions look similar.

Open-source tools

• SHAP
• LIME
• Captum
• InterpretML
• scikit-learn
• PyTorch
• TensorFlow
• MLflow
• DVC
• CycloneDX CLI
• MkDocs

Source-available / open-weight tools

• Alibi Explain (Business Source License 1.1; source-available)

Commercial / hosted tools

• Fiddler AI Observability
• Arize AX
• Evidently Cloud
• NannyML Cloud
• Amazon SageMaker Clarify
• Azure Machine Learning Responsible AI dashboard
• IBM watsonx.governance

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Lack of Explainability in Security AI Agents (L6)

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

• N/A

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Evaluation 6.3: Lack of Interpretability and Explainability