Scope boundary

Owns trusted reference artifacts for general predictive quality, serving latency and throughput, confidence or score distributions, and non-LLM output distributions under declared normal conditions.

AID-M-003.005: AID-M-003.002 baselines general predictive and operational metrics; AID-M-003.005 baselines reconstruction error for a governed image or media domain. A modality-specific reconstruction distribution should not be folded into a generic model KPI baseline.

AID-M-003.006: AID-M-003.002 owns general scalar performance and operational references; AID-M-003.006 owns graph snapshots, structural metrics, deterministic node ordering, and GNN probe outputs. Graph topology and probe-node behavior require graph-bound populations and cannot be represented accurately by generic KPI baselines alone.

AID-M-003.007: AID-M-003.002 owns general predictive and operational references for non-LLM behavior; AID-M-003.007 owns prompt-suite, refusal, policy-conformance, leakage, jailbreak, and optional embedding-distribution references for LLM applications. LLM behavioral baselines must remain bound to prompts, policies, tools, retrieval configuration, and evaluators rather than only generic model metrics.

AID-D-002: AID-M-003.002 creates and recalibrates the trusted normal reference; AID-D-002 compares live model behavior and performance against an approved reference to detect drift or anomalies. Baseline generation is not a production finding, and production monitoring must not silently redefine its own reference.

Implementation guidance

• AID-M-003.002-G001: Calculate and store key performance metrics on a trusted validation dataset.
• AID-M-003.002-G002: Establish baselines for operational metrics like latency and throughput via load testing.
• AID-M-003.002-G003: Baseline the model's output distribution on normal data.
• AID-M-003.002-G004: Link performance and operational baselines to specific model versions in a central model registry.

AID-M-003.002-G001

Concept:

Generate the reference only from a registry-approved model digest and a frozen validation set. Do not load pickle or joblib objects. This example uses skops with a deny-by-default type policy whose own digest is injected by the promotion control plane; any type not already trusted by skops must also appear in that reviewed policy.

Step 1: Pin the restricted skops type policy

# File: security/approved_skops_types.json
[]

from __future__ import annotations

# File: modeling/secure_model_load.py

import hashlib
import json
import os
import re
from pathlib import Path
from typing import Any

import skops.io as sio

SHA256 = re.compile(r"^[0-9a-f]{64}$")
TYPE_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.]*$")

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 load_approved_skops_model(model_path: Path) -> Any:
    if model_path.suffix != ".skops" or not model_path.is_file():
        raise SystemExit("model must be an existing .skops artifact")

    expected_model_sha = os.environ.get("MODEL_SHA256", "").lower()
    if not SHA256.fullmatch(expected_model_sha) or sha256_file(model_path) != expected_model_sha:
        raise SystemExit("model does not match the registry-approved MODEL_SHA256")

    policy_path = Path("security/approved_skops_types.json")
    expected_policy_sha = os.environ.get("APPROVED_SKOPS_POLICY_SHA256", "").lower()
    if not SHA256.fullmatch(expected_policy_sha) or sha256_file(policy_path) != expected_policy_sha:
        raise SystemExit("skops type policy does not match the protected approval digest")

    approved = json.loads(policy_path.read_text(encoding="utf-8"))
    if (
        not isinstance(approved, list)
        or any(not isinstance(name, str) or not TYPE_NAME.fullmatch(name) for name in approved)
        or len(approved) != len(set(approved))
    ):
        raise SystemExit("approved skops types must be a unique list of qualified type names")

    discovered = set(sio.get_untrusted_types(file=model_path))
    unauthorized = sorted(discovered - set(approved))
    if unauthorized:
        raise SystemExit("model contains unapproved skops types: " + ", ".join(unauthorized))
    return sio.load(model_path, trusted=sorted(discovered))

Step 2: Compute a version-bound binary-classification baseline

from __future__ import annotations

# File: modeling/calculate_performance_baseline.py

import json
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score

from modeling.secure_model_load import load_approved_skops_model, sha256_file

MODEL_PATH = Path("artifacts/model-v17.skops")
FEATURES_PATH = Path("data/X_val.csv")
LABELS_PATH = Path("data/y_val.csv")
OUTPUT_PATH = Path("baselines/model-v17-performance.json")

model = load_approved_skops_model(MODEL_PATH)
features = pd.read_csv(FEATURES_PATH)
labels_frame = pd.read_csv(LABELS_PATH)
if labels_frame.shape[1] != 1 or features.empty or len(features) != len(labels_frame):
    raise SystemExit("validation features and one-column labels must be non-empty and row-aligned")
feature_values = features.to_numpy(dtype=np.float64)
if not np.isfinite(feature_values).all():
    raise SystemExit("validation features contain NaN or infinity")
labels = labels_frame.iloc[:, 0].to_numpy()
if pd.isna(labels).any() or not set(np.unique(labels)).issubset({0, 1}):
    raise SystemExit("this baseline profile requires binary labels encoded as 0 and 1")

predictions = np.asarray(model.predict(features))
if predictions.shape != labels.shape or pd.isna(predictions).any():
    raise SystemExit("model predictions are missing or not row-aligned")
if not set(np.unique(predictions)).issubset({0, 1}):
    raise SystemExit("this baseline profile requires binary predictions encoded as 0 and 1")

metrics = {
    "accuracy": float(accuracy_score(labels, predictions)),
    "precision": float(precision_score(labels, predictions, zero_division=0)),
    "recall": float(recall_score(labels, predictions, zero_division=0)),
    "f1_score": float(f1_score(labels, predictions, zero_division=0)),
}
if not all(np.isfinite(value) and 0.0 <= value <= 1.0 for value in metrics.values()):
    raise SystemExit("calculated metric is outside the valid range")

baseline = {
    "schema_version": "aidefend.performance_baseline.v1",
    "model_sha256": sha256_file(MODEL_PATH),
    "features_sha256": sha256_file(FEATURES_PATH),
    "labels_sha256": sha256_file(LABELS_PATH),
    "sample_count": int(len(labels)),
    "created_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
    "metrics": metrics,
}
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_text(
    json.dumps(baseline, sort_keys=True, separators=(",", ":")),
    encoding="utf-8",
)

Step 3: Sign and promote the immutable reference

set -euo pipefail
python modeling/calculate_performance_baseline.py
cosign sign-blob --yes --key keys/baseline-signing.key --bundle baselines/model-v17-performance.sig baselines/model-v17-performance.json

Action: Promotion must inject the approved model and type-policy digests, verify the baseline signature, and reject missing or mismatched evidence. A baseline is a reference artifact; Detect-side controls decide whether later behavior is anomalous.

AID-M-003.002-G002

Concept

An operational baseline must bind the exact signed scenario, target origin, model digest, and deployment revision. A redirect, final-target drift, empty identity, or malformed release identity invalidates the run; its responses may remain diagnostic evidence but cannot contribute accepted latency or throughput measurements or a PASS.

Drive Locust from a signed scenario

The closed JSON scenario contains schema_version, a bounded scenario_id, a lowercase 64-hex model_digest, a deployment revision matching the explicit [A-Za-z0-9][A-Za-z0-9._-]{0,127} schema, an HTTPS origin-only host, an origin-relative endpoint_path, policy-owned load and timing values, and a non-empty population of unique case_id/body objects. The framework does not define universal load or latency thresholds.

# File: load_tests/locustfile.py
from __future__ import annotations

import itertools
import json
import os
import re
from pathlib import Path
from urllib.parse import urlparse

from locust import HttpUser, between, task

SCENARIO_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
MODEL_DIGEST = re.compile(r"^[0-9a-f]{64}$")
DEPLOYMENT_REVISION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")


def canonical_https_origin(raw: str) -> str:
    if not isinstance(raw, str) or not raw:
        raise RuntimeError("load target origin is empty")
    parsed = urlparse(raw)
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username
        or parsed.password
        or parsed.path not in ("", "/")
        or parsed.params
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("load host must be one credential-free HTTPS origin")
    try:
        port = parsed.port
    except ValueError as error:
        raise RuntimeError("load target port is invalid") from error
    hostname = parsed.hostname.lower()
    if ":" in hostname:
        hostname = f"[{hostname}]"
    return f"https://{hostname}" + (f":{port}" if port not in (None, 443) else "")


scenario = json.loads(Path(os.environ["LOAD_SCENARIO"]).read_text(encoding="utf-8"))
required = {
    "schema_version", "scenario_id", "model_digest", "deployment_revision",
    "host", "endpoint_path", "users", "spawn_rate", "run_time",
    "stop_timeout_seconds", "request_timeout_seconds",
    "wait_time_min_seconds", "wait_time_max_seconds", "payloads",
}
if not isinstance(scenario, dict) or set(scenario) != required:
    raise RuntimeError("load scenario fields differ")
if scenario["schema_version"] != "aidefend.operational-load-scenario.v1":
    raise RuntimeError("load scenario schema differs")
for field, pattern in {
    "scenario_id": SCENARIO_ID,
    "model_digest": MODEL_DIGEST,
    "deployment_revision": DEPLOYMENT_REVISION,
}.items():
    value = scenario[field]
    if not isinstance(value, str) or not value or not pattern.fullmatch(value):
        raise RuntimeError(f"load identity is invalid: {field}")
target_origin = canonical_https_origin(scenario["host"])
if scenario["host"] != target_origin:
    raise RuntimeError("load host must equal its canonical HTTPS origin")
endpoint_path = scenario["endpoint_path"]
if (
    not isinstance(endpoint_path, str)
    or not endpoint_path.startswith("/")
    or endpoint_path.startswith("//")
):
    raise RuntimeError("load endpoint must be an origin-relative path")
parsed_endpoint = urlparse(endpoint_path)
if parsed_endpoint.scheme or parsed_endpoint.netloc or parsed_endpoint.fragment:
    raise RuntimeError("load endpoint escaped the signed target origin")
target_path_query = (parsed_endpoint.path, parsed_endpoint.query)
target_url = target_origin + endpoint_path


def response_rejection(status_code: int, response_url: str) -> str | None:
    if isinstance(status_code, bool) or not isinstance(status_code, int):
        return "response status is invalid"
    if 300 <= status_code < 400:
        return f"redirect response rejected: {status_code}"
    try:
        observed = urlparse(response_url)
        observed_origin = canonical_https_origin(
            f"{observed.scheme}://{observed.netloc}"
        )
    except (RuntimeError, ValueError):
        return "response target is invalid"
    if (
        observed_origin != target_origin
        or (observed.path, observed.query) != target_path_query
        or observed.fragment
    ):
        return "response final target differs from signed target"
    if status_code < 200 or status_code >= 300:
        return f"unexpected HTTP status {status_code}"
    return None


if not isinstance(scenario["payloads"], list) or not scenario["payloads"]:
    raise RuntimeError("load payload population is empty")
if any(
    not isinstance(item, dict)
    or set(item) != {"case_id", "body"}
    or not isinstance(item["case_id"], str)
    or not __import__("re").fullmatch(r"[A-Za-z0-9._-]{1,64}", item["case_id"])
    or not isinstance(item["body"], dict)
    for item in scenario["payloads"]
):
    raise RuntimeError("every load case must have a bounded case_id and JSON body")
case_ids = [item["case_id"] for item in scenario["payloads"]]
if len(case_ids) != len(set(case_ids)):
    raise RuntimeError("load case population is duplicated")
timeout = scenario["request_timeout_seconds"]
wait_min = scenario["wait_time_min_seconds"]
wait_max = scenario["wait_time_max_seconds"]
for name, value in {
    "request_timeout_seconds": timeout,
    "wait_time_min_seconds": wait_min,
    "wait_time_max_seconds": wait_max,
}.items():
    if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0:
        raise RuntimeError(f"load timing value is invalid: {name}")
if timeout <= 0 or wait_max < wait_min:
    raise RuntimeError("load timing bounds are invalid")


class ModelAPIUser(HttpUser):
    wait_time = between(wait_min, wait_max)

    def on_start(self):
        self._payloads = itertools.cycle(scenario["payloads"])

    @task
    def predict_endpoint(self):
        case = next(self._payloads)
        with self.client.post(
            scenario["endpoint_path"],
            json=case["body"],
            timeout=timeout,
            allow_redirects=False,
            catch_response=True,
            name=f"policy_population/{case['case_id']}",
        ) as response:
            rejection = response_rejection(response.status_code, response.url)
            if rejection is not None:
                response.failure(rejection)

Run, summarize, and sign the complete result

# File: load_tests/run_baseline.sh
#!/usr/bin/env bash
set -euo pipefail
umask 077

test -n "$LOAD_SCENARIO"
test -n "$LOAD_SCENARIO_BUNDLE"
test -n "$LOAD_SCENARIO_VERIFY_KEY"
test -n "$LOAD_RESULT_SIGNING_KEY"
mkdir -p out/load
scenario_dir="$(mktemp -d)"
trap 'rm -rf -- "$scenario_dir"' EXIT
scenario_snapshot="$scenario_dir/scenario.json"
install -m 0400 -- "$LOAD_SCENARIO" "$scenario_snapshot"
cosign verify-blob --key "$LOAD_SCENARIO_VERIFY_KEY"   --bundle "$LOAD_SCENARIO_BUNDLE" "$scenario_snapshot" >/dev/null
LOAD_SCENARIO="$scenario_snapshot"

jq -e '
  .schema_version == "aidefend.operational-load-scenario.v1"
  and (.scenario_id | type == "string" and test("^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"))
  and (.model_digest | type == "string" and test("^[0-9a-f]{64}$"))
  and (.deployment_revision | type == "string" and test("^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"))
  and (.host | type == "string" and test("^https://([A-Za-z0-9.-]+|\\[[0-9A-Fa-f:]+\\])(:[1-9][0-9]{0,4})?$"))
  and (.users | type == "number" and floor == . and . > 0)
  and (.spawn_rate | type == "number" and . > 0)
  and (.run_time | type == "string" and test("^[1-9][0-9]*(s|m|h)$"))
  and (.stop_timeout_seconds | type == "number" and . > 0)
  and (.request_timeout_seconds | type == "number" and . > 0)
  and (.wait_time_min_seconds | type == "number" and . >= 0)
  and ((.wait_time_max_seconds | type) == "number" and .wait_time_max_seconds >= .wait_time_min_seconds)
  and (.payloads | type == "array" and length > 0)
' "$LOAD_SCENARIO" >/dev/null

export LOAD_SCENARIO
scenario_sha256="$(sha256sum -- "$LOAD_SCENARIO" | awk '{print $1}')"
scenario_id="$(jq -r .scenario_id "$LOAD_SCENARIO")"
model_digest="$(jq -r .model_digest "$LOAD_SCENARIO")"
deployment_revision="$(jq -r .deployment_revision "$LOAD_SCENARIO")"
host="$(jq -r .host "$LOAD_SCENARIO")"
users="$(jq -r .users "$LOAD_SCENARIO")"
spawn_rate="$(jq -r .spawn_rate "$LOAD_SCENARIO")"
run_time="$(jq -r .run_time "$LOAD_SCENARIO")"
stop_timeout="$(jq -r .stop_timeout_seconds "$LOAD_SCENARIO")"

set +e
locust -f load_tests/locustfile.py --headless --only-summary   --exit-code-on-error 2 --csv out/load/baseline --csv-full-history   --host "$host" --users "$users" --spawn-rate "$spawn_rate"   --run-time "$run_time" --stop-timeout "$stop_timeout"
locust_rc=$?
set -e
case "$locust_rc" in
  0|2) ;;
  *) echo "Locust execution error: $locust_rc" >&2; exit "$locust_rc" ;;
esac

python load_tests/summarize_locust.py   --scenario "$LOAD_SCENARIO"   --stats out/load/baseline_stats.csv   --failures out/load/baseline_failures.csv   --locust-exit-code "$locust_rc"   --output out/load/operational-baseline.json

jq -e \
  --arg scenario_sha256 "$scenario_sha256" \
  --arg scenario_id "$scenario_id" \
  --arg model_digest "$model_digest" \
  --arg deployment_revision "$deployment_revision" \
  --arg target_origin "$host" \
  --argjson locust_rc "$locust_rc" '
  .scenario_sha256 == $scenario_sha256
  and .scenario_id == $scenario_id
  and .model_digest == $model_digest
  and .deployment_revision == $deployment_revision
  and .target_origin == $target_origin
  and .locust_exit_code == $locust_rc
  and (.outcome == "PASS" or .outcome == "FAIL")
  and (.measurement_accepted == (.outcome == "PASS"))
  and .request_count > 0
' out/load/operational-baseline.json >/dev/null
cosign sign-blob --yes --key "$LOAD_RESULT_SIGNING_KEY"   --bundle out/load/operational-baseline.sigstore.json   out/load/operational-baseline.json
outcome="$(jq -r .outcome out/load/operational-baseline.json)"
[[ "$outcome" == "PASS" ]] || exit 1

# File: load_tests/summarize_locust.py
from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
from pathlib import Path
from urllib.parse import urlparse

parser = argparse.ArgumentParser()
parser.add_argument("--scenario", required=True)
parser.add_argument("--stats", required=True)
parser.add_argument("--failures", required=True)
parser.add_argument("--locust-exit-code", type=int, required=True)
parser.add_argument("--output", required=True)
args = parser.parse_args()

scenario_raw = Path(args.scenario).read_bytes()
scenario = json.loads(scenario_raw)
SCENARIO_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
MODEL_DIGEST = re.compile(r"^[0-9a-f]{64}$")
DEPLOYMENT_REVISION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
for field, pattern in {
    "scenario_id": SCENARIO_ID,
    "model_digest": MODEL_DIGEST,
    "deployment_revision": DEPLOYMENT_REVISION,
}.items():
    value = scenario.get(field)
    if not isinstance(value, str) or not value or not pattern.fullmatch(value):
        raise SystemExit(f"scenario identity is invalid: {field}")
host = scenario.get("host")
parsed_host = urlparse(host if isinstance(host, str) else "")
if (
    parsed_host.scheme != "https"
    or not parsed_host.hostname
    or parsed_host.username
    or parsed_host.password
    or parsed_host.path
    or parsed_host.params
    or parsed_host.query
    or parsed_host.fragment
):
    raise SystemExit("scenario host is not an exact HTTPS origin")
try:
    host_port = parsed_host.port
except ValueError as error:
    raise SystemExit("scenario host port is invalid") from error
host_name = parsed_host.hostname.lower()
if ":" in host_name:
    host_name = f"[{host_name}]"
canonical_host = (
    f"https://{host_name}"
    + (f":{host_port}" if host_port not in (None, 443) else "")
)
if host != canonical_host:
    raise SystemExit("scenario host is not canonical")
endpoint_path = scenario.get("endpoint_path")
if not isinstance(endpoint_path, str) or not endpoint_path.startswith("/") or endpoint_path.startswith("//"):
    raise SystemExit("scenario endpoint path is invalid")
parsed_endpoint = urlparse(endpoint_path)
if parsed_endpoint.scheme or parsed_endpoint.netloc or parsed_endpoint.fragment:
    raise SystemExit("scenario endpoint escaped its target origin")
target_origin = host
target_url = target_origin + endpoint_path
with Path(args.stats).open(newline="", encoding="utf-8") as handle:
    rows = list(csv.DictReader(handle))
aggregate = [row for row in rows if row.get("Name") == "Aggregated"]
if len(aggregate) != 1:
    raise SystemExit("Locust aggregate population differs")
row = aggregate[0]
with Path(args.failures).open(newline="", encoding="utf-8") as handle:
    failure_rows = list(csv.DictReader(handle))

request_count = int(row["Request Count"])
failure_count = int(row["Failure Count"])
expected_case_ids = {item["case_id"] for item in scenario["payloads"]}
observed_case_ids = {
    item["Name"].removeprefix("policy_population/")
    for item in rows
    if item.get("Name", "").startswith("policy_population/")
    and int(item["Request Count"]) > 0
}
missing_case_ids = sorted(expected_case_ids - observed_case_ids)
unexpected_case_ids = sorted(observed_case_ids - expected_case_ids)
population_complete = (
    request_count > 0
    and not missing_case_ids
    and not unexpected_case_ids
)
measurement_accepted = (
    population_complete
    and failure_count == 0
    and not failure_rows
    and args.locust_exit_code == 0
)
report = {
    "schema_version": "aidefend.operational-baseline.v1",
    "scenario_id": scenario["scenario_id"],
    "scenario_sha256": hashlib.sha256(scenario_raw).hexdigest(),
    "model_digest": scenario["model_digest"],
    "deployment_revision": scenario["deployment_revision"],
    "target_origin": target_origin,
    "target_url": target_url,
    "request_count": request_count,
    "failure_count": failure_count,
    "failure_kind_count": len(failure_rows),
    "expected_case_count": len(expected_case_ids),
    "observed_case_count": len(observed_case_ids),
    "missing_case_ids": missing_case_ids,
    "unexpected_case_ids": unexpected_case_ids,
    "average_response_time_ms": (
        float(row["Average Response Time"]) if measurement_accepted else None
    ),
    "p95_response_time_ms": (
        float(row["95%"]) if measurement_accepted else None
    ),
    "requests_per_second": (
        float(row["Requests/s"]) if measurement_accepted else None
    ),
    "locust_exit_code": args.locust_exit_code,
    "population_complete": population_complete,
    "measurement_accepted": measurement_accepted,
}
report["outcome"] = "PASS" if measurement_accepted else "FAIL"
Path(args.output).write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")

Fail-closed fixtures

Set each of scenario_id, model_digest, and deployment_revision to null, an empty string, and an out-of-schema value; each scenario must stop before Locust starts. Return a 302 with a valid Location header and separately inject a response whose final URL has another origin or path; response_rejection must mark both as failures, the signed report must set measurement_accepted=false, and latency/throughput fields must be null. A direct 2xx response from the exact signed target remains the positive fixture.

Action: Promote the baseline only when the scenario signature verifies, Locust exits successfully, at least one request is measured, every signed case is exercised, no redirect or target drift occurs, and the signed report binds the exact canonical target origin and URL, scenario ID/digest, 64-hex model digest, deployment revision, and Locust exit code. Retain signed FAIL evidence, but never use its null measurement fields as a baseline. Run the policy-owned profile enough times to characterize warm-up and steady state without averaging missing, redirected, drifted, or failed populations away.

AID-M-003.002-G003

Concept:

Generate output-distribution evidence with the same registry-approved .skops model and frozen feature set used by the performance baseline. Write a separate signed artifact rather than mutating an earlier baseline file, so its model, input, and sample bindings remain independently verifiable.

from __future__ import annotations

# File: modeling/calculate_distribution_baseline.py

import json
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd

from modeling.secure_model_load import load_approved_skops_model, sha256_file

MODEL_PATH = Path("artifacts/model-v17.skops")
FEATURES_PATH = Path("data/X_val.csv")
OUTPUT_PATH = Path("baselines/model-v17-output-distribution.json")

model = load_approved_skops_model(MODEL_PATH)
features = pd.read_csv(FEATURES_PATH)
if features.empty:
    raise SystemExit("golden feature set is empty")
values = features.to_numpy(dtype=np.float64)
if not np.isfinite(values).all():
    raise SystemExit("golden feature set contains NaN or infinity")

predictions = np.asarray(model.predict(features))
if predictions.shape != (len(features),) or pd.isna(predictions).any():
    raise SystemExit("predictions are missing or not row-aligned")
if not set(np.unique(predictions)).issubset({0, 1}):
    raise SystemExit("this distribution profile requires binary predictions encoded as 0 and 1")

counts = {str(label): int(np.sum(predictions == label)) for label in (0, 1)}
sample_count = int(len(predictions))
distribution = {
    label: {
        "count": count,
        "rate": float(count / sample_count),
    }
    for label, count in counts.items()
}
if sum(item["count"] for item in distribution.values()) != sample_count:
    raise SystemExit("distribution counts do not cover every prediction")

artifact = {
    "schema_version": "aidefend.output_distribution_baseline.v1",
    "model_sha256": sha256_file(MODEL_PATH),
    "features_sha256": sha256_file(FEATURES_PATH),
    "sample_count": sample_count,
    "created_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
    "classes": distribution,
}
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUTPUT_PATH.write_text(
    json.dumps(artifact, sort_keys=True, separators=(",", ":")),
    encoding="utf-8",
)

set -euo pipefail
python modeling/calculate_distribution_baseline.py
cosign sign-blob --yes --key keys/baseline-signing.key --bundle baselines/model-v17-output-distribution.sig baselines/model-v17-output-distribution.json

Action: Register the signed distribution URI and SHA-256 beside the exact model version. Detect-side comparison jobs must reject a missing signature, model mismatch, input mismatch, or floating baseline URI before using the reference.

AID-M-003.002-G004

Concept

A model-version tag can help responders discover a baseline, but it is mutable metadata and cannot make an artifact immutable. Bind the exact registered version to a signature-verified baseline manifest, independently resolve and hash the referenced bytes, then use MLflow tags only as read-back-checked pointers.

Verify the baseline before writing the registry link

The signed manifest contains exactly schema_version, policy_version, tracking_uri, model_name, model_version, model_source_run_id, baseline_uri, baseline_sha256, baseline_type, and release_manifest_sha256. The artifact resolver and registry endpoint are authenticated platform dependencies; the CI identity must not be able to rewrite the versioned baseline object.

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

import hashlib
import json
import os
import re
import subprocess
import tempfile
from pathlib import Path
from urllib.parse import urlparse

from mlflow import MlflowClient
from mlflow.artifacts import download_artifacts

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


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


def regular_bytes(path: Path) -> bytes:
    if path.is_symlink() or not path.is_file():
        raise ValueError(f"baseline input is not a regular file: {path}")
    raw = path.read_bytes()
    if not raw:
        raise ValueError("baseline input is empty")
    return raw


manifest_path = Path(os.environ["BASELINE_BINDING_MANIFEST"])
bundle_path = Path(os.environ["BASELINE_BINDING_BUNDLE"])
verify_key = os.environ["BASELINE_BINDING_VERIFY_KEY"]
verify_timeout = float(os.environ["BASELINE_BINDING_VERIFY_TIMEOUT_SECONDS"])
manifest_raw = regular_bytes(manifest_path)
bundle_raw = regular_bytes(bundle_path)
subprocess.run(
    ["cosign", "verify-blob", "--key", verify_key, "--bundle", str(bundle_path), str(manifest_path)],
    check=True, capture_output=True, text=True, timeout=verify_timeout,
)
if regular_bytes(manifest_path) != manifest_raw or regular_bytes(bundle_path) != bundle_raw:
    raise RuntimeError("baseline binding inputs changed during verification")

manifest = json.loads(
    manifest_raw.decode("utf-8", errors="strict"),
    object_pairs_hook=unique_object,
)
required = {
    "schema_version", "policy_version", "tracking_uri", "model_name",
    "model_version", "model_source_run_id", "baseline_uri",
    "baseline_sha256", "baseline_type", "release_manifest_sha256",
}
if not isinstance(manifest, dict) or set(manifest) != required:
    raise ValueError("baseline binding manifest fields differ")
if manifest["schema_version"] != "aidefend.model-baseline-binding.v1":
    raise ValueError("baseline binding manifest schema differs")
if any(not isinstance(manifest[key], str) or not manifest[key] for key in required):
    raise ValueError("baseline binding manifest contains an empty binding")
if not SHA256.fullmatch(manifest["baseline_sha256"]) or not SHA256.fullmatch(
    manifest["release_manifest_sha256"]
):
    raise ValueError("baseline or release-manifest digest is invalid")
if urlparse(manifest["tracking_uri"]).scheme in {"", "http"}:
    raise ValueError("production MLflow endpoint must use authenticated secure transport")
if urlparse(manifest["baseline_uri"]).scheme in {"", "http"}:
    raise ValueError("baseline URI must use an authenticated immutable artifact resolver")

with tempfile.TemporaryDirectory(prefix="aidefend-baseline-") as directory:
    local = Path(download_artifacts(artifact_uri=manifest["baseline_uri"], dst_path=directory))
    baseline_raw = regular_bytes(local)
if hashlib.sha256(baseline_raw).hexdigest() != manifest["baseline_sha256"]:
    raise ValueError("resolved baseline bytes do not match the signed manifest")

client = MlflowClient(tracking_uri=manifest["tracking_uri"])
version = client.get_model_version(manifest["model_name"], manifest["model_version"])
if version.run_id != manifest["model_source_run_id"]:
    raise ValueError("model version source run differs from the signed binding")

manifest_sha256 = hashlib.sha256(manifest_raw).hexdigest()
tags = {
    "aidefend.baseline_uri": manifest["baseline_uri"],
    "aidefend.baseline_sha256": manifest["baseline_sha256"],
    "aidefend.baseline_type": manifest["baseline_type"],
    "aidefend.baseline_binding_manifest_sha256": manifest_sha256,
    "aidefend.release_manifest_sha256": manifest["release_manifest_sha256"],
}
for key, value in tags.items():
    client.set_model_version_tag(manifest["model_name"], manifest["model_version"], key, value)

readback = client.get_model_version(manifest["model_name"], manifest["model_version"])
if readback.run_id != manifest["model_source_run_id"] or any(
    readback.tags.get(key) != value for key, value in tags.items()
):
    raise RuntimeError("MLflow baseline pointer readback differs")

print(json.dumps({
    "schema_version": "aidefend.model-baseline-binding-result.v1",
    "model_name": manifest["model_name"],
    "model_version": manifest["model_version"],
    "model_source_run_id": manifest["model_source_run_id"],
    "baseline_sha256": manifest["baseline_sha256"],
    "binding_manifest_sha256": manifest_sha256,
    "tag_readback": "matched",
}, sort_keys=True))

Action: The promotion controller must verify the signed manifest and baseline bytes again before assigning its policy-defined deployment alias. Do not treat the mutable tags, an unvalidated URI, or caller-supplied digest text as the baseline identity.

Open-source tools

• scikit-learn
• skops
• MLflow
• DVC
• Evidently
• NannyML
• Locust
• Grafana k6
• Apache JMeter
• Prometheus
• Grafana

Source-available / open-weight tools

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

Commercial / hosted tools

• Arize AX
• Fiddler AI Observability
• Evidently Cloud
• NannyML Cloud
• Amazon SageMaker Model Monitor
• Google Vertex AI Model Monitoring
• Azure Machine Learning model monitoring
• Datadog
• New Relic

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

• N/A

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Evaluation 6.2: Insufficient evaluation data
• Algorithms 5.1: Lack of tracking and reproducibility of experiments