Scope boundary

Owns classical-plus-post-quantum provenance and integrity continuity for long-lived AI assets whose trust must survive cryptographic migration, including hybrid signatures, algorithm policy, verification history, and migration evidence.

AID-M-002.002: AID-M-002.005 preserves integrity trust across a long-lived classical-to-post-quantum migration; AID-M-002.002 performs ordinary whole-artifact digest and signature verification. Use the hybrid control only when cryptographic longevity and transition continuity are in scope; it does not replace routine artifact verification.

AID-H-003.008: AID-M-002.005 records and verifies long-lived hybrid provenance continuity; AID-H-003.008 enforces crypto-agile signing and distribution requirements when artifacts are admitted or delivered. The Model control establishes durable trust evidence, while the Harden control blocks distribution or admission that does not meet the active cryptographic policy.

Warning

Medium on signature size, archival overhead, and verifier interoperability

ML-DSA signatures are materially larger than classical signatures, and every verifier in the restore path must be tested against the selected parameter set. During migration, require both the classical and ML-DSA signatures; do not silently accept either one alone unless an explicitly versioned emergency policy says so.

Implementation guidance

• AID-M-002.005-G001: Issue and verify separate classical cosign and ML-DSA signatures over one canonical, expiring provenance statement.
• AID-M-002.005-G002: Re-verify and roll long-lived provenance before expiry or trust-policy change, preserving an auditable continuity chain for restore.

AID-M-002.005-G001

Concept:

The approved object is not just a signature pair. A protected approval workflow signs a local verifier policy that names the exact current statement digest and chain root. The restore verifier also requires the policy digest injected by the deployment control plane, which prevents an archived statement or an older still-valid policy from selecting itself.

Step 1: Create one canonical statement and immutable chain node

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")
STATEMENT_LIFETIME_SECONDS = _aidefend_positive_int("M002005_STATEMENT_LIFETIME_SECONDS")

# File: archive/create_hybrid_statement.py

import hashlib
import json
import os
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path

ASSET_ID = "model-credit-risk-v12"
BASE = Path("archive")
ARTIFACT = BASE / "model-credit-risk-v12.safetensors"
PROVENANCE = BASE / "attestations/model-credit-risk-v12.intoto.jsonl"
CURRENT = BASE / "model-credit-risk-v12.provenance.json"
CHAIN_DIR = BASE / "statements"
SHA256 = re.compile(r"^[0-9a-f]{64}$")
KMS_ARN = re.compile(r"^arn:aws(?:-us-gov|-cn)?:kms:[a-z0-9-]+:[0-9]{12}:key/[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 canonical(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

for required in (ARTIFACT, PROVENANCE):
    if not required.is_file():
        raise SystemExit(f"required archive input missing: {required}")

pqc_key_arn = os.environ.get("ML_DSA_KMS_KEY_ARN", "")
if not KMS_ARN.fullmatch(pqc_key_arn):
    raise SystemExit("ML_DSA_KMS_KEY_ARN must be one resolved AWS KMS key ARN")

previous = os.environ.get("PREVIOUS_STATEMENT_SHA256") or None
if previous is not None:
    previous = previous.lower()
    if not SHA256.fullmatch(previous):
        raise SystemExit("PREVIOUS_STATEMENT_SHA256 must be a lowercase SHA-256")

now = datetime.now(timezone.utc).replace(microsecond=0)
statement = {
    "schema_version": "aidefend.long_lived_provenance.v1",
    "asset_id": ASSET_ID,
    "artifact_uri": "s3://ai-archive/model-credit-risk-v12.safetensors",
    "artifact_sha256": sha256_file(ARTIFACT),
    "provenance_uri": "s3://ai-archive/attestations/model-credit-risk-v12.intoto.jsonl",
    "provenance_sha256": sha256_file(PROVENANCE),
    "created_at": now.isoformat().replace("+00:00", "Z"),
    "expires_at": (now + timedelta(seconds=STATEMENT_LIFETIME_SECONDS)).isoformat().replace("+00:00", "Z"),
    "classical_profile": "cosign-ecdsa-p256",
    "pqc_profile": "ML_DSA_65/ML_DSA_SHAKE_256",
    "pqc_key_arn": pqc_key_arn,
    "previous_statement_sha256": previous,
}
encoded = canonical(statement)
if len(encoded) > 4096:
    raise SystemExit("canonical statement exceeds the AWS KMS RAW message limit")

BASE.mkdir(parents=True, exist_ok=True)
CHAIN_DIR.mkdir(parents=True, exist_ok=True)
temporary = CURRENT.with_suffix(".json.tmp")
temporary.write_bytes(encoded)
temporary.replace(CURRENT)
statement_sha256 = hashlib.sha256(encoded).hexdigest()
chain_node = CHAIN_DIR / f"{statement_sha256}.json"
if chain_node.exists() and chain_node.read_bytes() != encoded:
    raise SystemExit("immutable statement digest path already contains different bytes")
chain_node.write_bytes(encoded)
print(statement_sha256)

Step 2: Resolve one KMS ARN and produce both statement signatures

set -euo pipefail
KMS_KEY_ARN="$(aws kms describe-key --key-id alias/aidefend-archive-mldsa --query KeyMetadata.Arn --output text)"
test -n "$KMS_KEY_ARN"
export ML_DSA_KMS_KEY_ARN="$KMS_KEY_ARN"
python archive/create_hybrid_statement.py
cosign sign-blob --yes --key keys/archive-classical.key --bundle archive/model-credit-risk-v12.cosign.sig archive/model-credit-risk-v12.provenance.json
aws kms sign --key-id "$KMS_KEY_ARN" --message fileb://archive/model-credit-risk-v12.provenance.json --message-type RAW --signing-algorithm ML_DSA_SHAKE_256 --query Signature --output text | base64 --decode > archive/model-credit-risk-v12.mldsa.sig
aws kms get-public-key --key-id "$KMS_KEY_ARN" --query PublicKey --output text | base64 --decode > archive/archive-mldsa-public.der

Step 3: Build and sign the verifier policy in the approval job

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")
VERIFIER_POLICY_LIFETIME_SECONDS = _aidefend_positive_int("M002005_VERIFIER_POLICY_LIFETIME_SECONDS")
MAX_STATEMENT_LIFETIME_SECONDS = _aidefend_positive_int("M002005_MAX_STATEMENT_LIFETIME_SECONDS")
MAX_FUTURE_SKEW_SECONDS = _aidefend_nonnegative_int("M002005_MAX_FUTURE_SKEW_SECONDS")

# File: archive/build_verifier_policy.py

import hashlib
import json
import os
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path

ASSET_ID = "model-credit-risk-v12"
BASE = Path("archive")
CURRENT = BASE / "model-credit-risk-v12.provenance.json"
CLASSICAL_PUBLIC_KEY = Path("trust/archive-classical.pub")
SHA256 = re.compile(r"^[0-9a-f]{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 canonical(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

raw_statement = CURRENT.read_bytes()
statement = json.loads(raw_statement)
if canonical(statement) != raw_statement:
    raise SystemExit("current statement is not canonical JSON")
current_head = hashlib.sha256(raw_statement).hexdigest()
chain_root = os.environ.get("CHAIN_ROOT_SHA256", current_head).lower()
if not SHA256.fullmatch(chain_root):
    raise SystemExit("CHAIN_ROOT_SHA256 must be a lowercase SHA-256")
if not (BASE / "statements" / f"{chain_root}.json").is_file():
    raise SystemExit("approved chain root is not present in the immutable statement store")

now = datetime.now(timezone.utc).replace(microsecond=0)
policy = {
    "schema_version": "aidefend.hybrid_verifier_policy.v1",
    "policy_version": os.environ.get("ARCHIVE_POLICY_VERSION", now.strftime("%Y%m%d%H%M%S")),
    "created_at": now.isoformat().replace("+00:00", "Z"),
    "expires_at": (now + timedelta(seconds=VERIFIER_POLICY_LIFETIME_SECONDS)).isoformat().replace("+00:00", "Z"),
    "expected_profiles": {
        "classical": "cosign-ecdsa-p256",
        "pqc": "ML_DSA_65/ML_DSA_SHAKE_256",
    },
    "max_statement_lifetime_seconds": MAX_STATEMENT_LIFETIME_SECONDS,
    "max_future_skew_seconds": MAX_FUTURE_SKEW_SECONDS,
    "classical_public_key": {
        "path": str(CLASSICAL_PUBLIC_KEY),
        "sha256": sha256_file(CLASSICAL_PUBLIC_KEY),
    },
    "trusted_pqc_key_arns": [statement["pqc_key_arn"]],
    "assets": {
        ASSET_ID: {
            "artifact_path": str(BASE / "model-credit-risk-v12.safetensors"),
            "artifact_uri": statement["artifact_uri"],
            "provenance_path": str(BASE / "attestations/model-credit-risk-v12.intoto.jsonl"),
            "provenance_uri": statement["provenance_uri"],
            "statement_path": str(CURRENT),
            "classical_signature_path": str(BASE / "model-credit-risk-v12.cosign.sig"),
            "pqc_signature_path": str(BASE / "model-credit-risk-v12.mldsa.sig"),
            "chain_directory": str(BASE / "statements"),
            "current_statement_sha256": current_head,
            "chain_root_sha256": chain_root,
        }
    },
}
Path("archive/verifier_policy.json").write_bytes(canonical(policy))

set -euo pipefail
export ARCHIVE_POLICY_VERSION="2026-07-10.1"
python archive/build_verifier_policy.py
cosign sign-blob --yes --key keys/verifier-policy-root.key --bundle archive/verifier_policy.cosign.sig archive/verifier_policy.json
sha256sum archive/verifier_policy.json
# Publish the printed digest as protected deployment variable APPROVED_ARCHIVE_POLICY_SHA256.

Step 4: Verify the protected policy, current head, chain, files, and both signatures

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")
MAX_STATEMENT_LIFETIME_SECONDS = _aidefend_positive_int("M002005_MAX_STATEMENT_LIFETIME_SECONDS")
MAX_FUTURE_SKEW_SECONDS = _aidefend_nonnegative_int("M002005_MAX_FUTURE_SKEW_SECONDS")
MAX_VERIFIER_POLICY_LIFETIME_SECONDS = _aidefend_positive_int("M002005_MAX_VERIFIER_POLICY_LIFETIME_SECONDS")
ARCHIVE_VERIFY_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("M002005_ARCHIVE_VERIFY_COMMAND_TIMEOUT_SECONDS")

# File: archive/verify_hybrid_statement.py

import hashlib
import json
import os
import re
import subprocess
from datetime import datetime, timedelta, timezone
from pathlib import Path

ASSET_ID = "model-credit-risk-v12"
POLICY_PATH = Path("archive/verifier_policy.json")
POLICY_SIGNATURE = Path("archive/verifier_policy.cosign.sig")
POLICY_ROOT_PUBLIC_KEY = Path("trust/verifier-policy-root.pub")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
STATEMENT_FIELDS = {
    "schema_version", "asset_id", "artifact_uri", "artifact_sha256",
    "provenance_uri", "provenance_sha256", "created_at", "expires_at",
    "classical_profile", "pqc_profile", "pqc_key_arn",
    "previous_statement_sha256",
}
POLICY_FIELDS = {
    "schema_version", "policy_version", "created_at", "expires_at",
    "expected_profiles", "max_statement_lifetime_seconds",
    "max_future_skew_seconds", "classical_public_key",
    "trusted_pqc_key_arns", "assets",
}
ASSET_POLICY_FIELDS = {
    "artifact_path", "artifact_uri", "provenance_path", "provenance_uri",
    "statement_path", "classical_signature_path", "pqc_signature_path",
    "chain_directory", "current_statement_sha256", "chain_root_sha256",
}

def fail(message: str) -> None:
    raise SystemExit(message)

def exact_keys(value: object, expected: set[str], label: str) -> dict:
    if not isinstance(value, dict) or set(value) != expected:
        fail(f"{label} fields do not match the approved schema")
    return value

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 canonical(value: dict) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")

def read_canonical(path: Path, label: str) -> tuple[dict, bytes]:
    raw = path.read_bytes()
    value = json.loads(raw)
    if canonical(value) != raw:
        fail(f"{label} is not canonical JSON")
    return value, raw

def parse_time(value: object, label: str) -> datetime:
    if not isinstance(value, str):
        fail(f"{label} must be an RFC 3339 timestamp")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        fail(f"{label} is not a valid RFC 3339 timestamp")
    if parsed.utcoffset() is None:
        fail(f"{label} must include a timezone")
    return parsed.astimezone(timezone.utc)

approved_policy_sha = os.environ.get("APPROVED_ARCHIVE_POLICY_SHA256", "").lower()
if not SHA256.fullmatch(approved_policy_sha):
    fail("protected APPROVED_ARCHIVE_POLICY_SHA256 is missing or invalid")
if sha256_file(POLICY_PATH) != approved_policy_sha:
    fail("local verifier policy is not the deployment-approved policy")
subprocess.run([
    "cosign", "verify-blob", "--key", str(POLICY_ROOT_PUBLIC_KEY),
    "--bundle", str(POLICY_SIGNATURE), str(POLICY_PATH),
], check=True, capture_output=True, text=True, timeout=ARCHIVE_VERIFY_COMMAND_TIMEOUT_SECONDS)

policy, _ = read_canonical(POLICY_PATH, "verifier policy")
exact_keys(policy, POLICY_FIELDS, "verifier policy")
if policy["schema_version"] != "aidefend.hybrid_verifier_policy.v1":
    fail("unsupported verifier policy schema")
if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
    fail("policy_version must be a non-empty string")
if policy["expected_profiles"] != {
    "classical": "cosign-ecdsa-p256",
    "pqc": "ML_DSA_65/ML_DSA_SHAKE_256",
}:
    fail("verifier policy profile mismatch")
if not isinstance(policy["max_statement_lifetime_seconds"], int) or not 1 <= policy["max_statement_lifetime_seconds"] <= MAX_STATEMENT_LIFETIME_SECONDS:
    fail("invalid maximum statement lifetime")
if not isinstance(policy["max_future_skew_seconds"], int) or not 0 <= policy["max_future_skew_seconds"] <= MAX_FUTURE_SKEW_SECONDS:
    fail("invalid future-skew allowance")

now = datetime.now(timezone.utc)
policy_created = parse_time(policy["created_at"], "policy created_at")
policy_expires = parse_time(policy["expires_at"], "policy expires_at")
if policy_created > now + timedelta(seconds=policy["max_future_skew_seconds"]):
    fail("verifier policy is future-dated")
if policy_expires <= now or policy_expires <= policy_created:
    fail("verifier policy is expired or has an invalid interval")
if policy_expires - policy_created > timedelta(seconds=MAX_VERIFIER_POLICY_LIFETIME_SECONDS):
    fail("verifier policy lifetime exceeds the runtime-profile maximum")

key_policy = exact_keys(policy["classical_public_key"], {"path", "sha256"}, "classical public key policy")
classical_public_key = Path(key_policy["path"])
if not SHA256.fullmatch(str(key_policy["sha256"])) or sha256_file(classical_public_key) != key_policy["sha256"]:
    fail("classical statement verification key digest mismatch")
trusted_pqc = policy["trusted_pqc_key_arns"]
if not isinstance(trusted_pqc, list) or not trusted_pqc or len(trusted_pqc) != len(set(trusted_pqc)):
    fail("trusted_pqc_key_arns must be a non-empty unique list")
if not isinstance(policy["assets"], dict) or set(policy["assets"]) != {ASSET_ID}:
    fail("policy must contain exactly the expected asset")
asset_policy = exact_keys(policy["assets"][ASSET_ID], ASSET_POLICY_FIELDS, "asset verifier policy")
for digest_field in ("current_statement_sha256", "chain_root_sha256"):
    if not SHA256.fullmatch(str(asset_policy[digest_field])):
        fail(f"{digest_field} is not a SHA-256")

statement_path = Path(asset_policy["statement_path"])
statement, current_raw = read_canonical(statement_path, "current statement")
exact_keys(statement, STATEMENT_FIELDS, "current statement")
if hashlib.sha256(current_raw).hexdigest() != asset_policy["current_statement_sha256"]:
    fail("current statement does not match the signed policy head")

def validate_statement(value: dict, label: str, require_current_key: bool) -> datetime:
    exact_keys(value, STATEMENT_FIELDS, label)
    if value["schema_version"] != "aidefend.long_lived_provenance.v1":
        fail(f"{label} has an unsupported schema")
    if value["asset_id"] != ASSET_ID:
        fail(f"{label} asset_id mismatch")
    if value["classical_profile"] != policy["expected_profiles"]["classical"] or value["pqc_profile"] != policy["expected_profiles"]["pqc"]:
        fail(f"{label} signature profile mismatch")
    if require_current_key and value["pqc_key_arn"] not in trusted_pqc:
        fail("current statement uses an untrusted ML-DSA key")
    for field in ("artifact_sha256", "provenance_sha256"):
        if not SHA256.fullmatch(str(value[field])):
            fail(f"{label} {field} is invalid")
    previous = value["previous_statement_sha256"]
    if previous is not None and not SHA256.fullmatch(str(previous)):
        fail(f"{label} previous statement digest is invalid")
    created = parse_time(value["created_at"], f"{label} created_at")
    expires = parse_time(value["expires_at"], f"{label} expires_at")
    if expires <= created:
        fail(f"{label} has an invalid validity interval")
    if expires - created > timedelta(seconds=policy["max_statement_lifetime_seconds"]):
        fail(f"{label} exceeds maximum statement lifetime")
    return created

current_created = validate_statement(statement, "current statement", True)
current_expires = parse_time(statement["expires_at"], "current statement expires_at")
if current_created > now + timedelta(seconds=policy["max_future_skew_seconds"]):
    fail("current statement is future-dated")
if current_expires <= now:
    fail("current statement is expired")
if statement["artifact_uri"] != asset_policy["artifact_uri"] or statement["provenance_uri"] != asset_policy["provenance_uri"]:
    fail("statement URI does not match signed policy")
if sha256_file(Path(asset_policy["artifact_path"])) != statement["artifact_sha256"]:
    fail("artifact digest mismatch")
if sha256_file(Path(asset_policy["provenance_path"])) != statement["provenance_sha256"]:
    fail("provenance digest mismatch")

subprocess.run([
    "cosign", "verify-blob", "--key", str(classical_public_key),
    "--bundle", asset_policy["classical_signature_path"], str(statement_path),
], check=True, capture_output=True, text=True, timeout=ARCHIVE_VERIFY_COMMAND_TIMEOUT_SECONDS)
kms_result = subprocess.run([
    "aws", "kms", "verify", "--key-id", statement["pqc_key_arn"],
    "--message", "fileb://" + str(statement_path), "--message-type", "RAW",
    "--signature", "fileb://" + asset_policy["pqc_signature_path"],
    "--signing-algorithm", "ML_DSA_SHAKE_256", "--output", "json",
], check=True, capture_output=True, text=True, timeout=ARCHIVE_VERIFY_COMMAND_TIMEOUT_SECONDS)
if json.loads(kms_result.stdout).get("SignatureValid") is not True:
    fail("ML-DSA signature invalid")

head = asset_policy["current_statement_sha256"]
root = asset_policy["chain_root_sha256"]
chain_directory = Path(asset_policy["chain_directory"])
expected = head
child_created: datetime | None = None
seen: set[str] = set()
chain_length = 0
while True:
    if expected in seen or len(seen) >= 4096:
        fail("statement chain contains a cycle or exceeds its bound")
    seen.add(expected)
    node_path = chain_directory / f"{expected}.json"
    node, node_raw = read_canonical(node_path, f"statement chain node {expected}")
    if hashlib.sha256(node_raw).hexdigest() != expected:
        fail("statement chain node digest does not match its filename")
    if expected == head and node_raw != current_raw:
        fail("current statement differs from its immutable chain node")
    node_created = validate_statement(node, f"statement chain node {expected}", False)
    if child_created is not None and node_created >= child_created:
        fail("statement chain timestamps are not strictly ordered")
    for field in ("asset_id", "artifact_uri", "artifact_sha256", "provenance_uri", "provenance_sha256"):
        if node[field] != statement[field]:
            fail(f"statement chain changes immutable asset binding: {field}")
    chain_length += 1
    previous = node["previous_statement_sha256"]
    if expected == root:
        if previous is not None:
            fail("signed chain root is not the first statement")
        break
    if previous is None:
        fail("statement chain ended before the signed chain root")
    child_created = node_created
    expected = previous

evidence = {
    "schema_version": "aidefend.hybrid_verification_result.v1",
    "asset_id": ASSET_ID,
    "verified_at": now.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
    "policy_sha256": approved_policy_sha,
    "policy_version": policy["policy_version"],
    "current_statement_sha256": head,
    "chain_root_sha256": root,
    "chain_length": chain_length,
    "artifact_sha256": statement["artifact_sha256"],
    "provenance_sha256": statement["provenance_sha256"],
    "classical_signature_verified": True,
    "mldsa_signature_verified": True,
}
result_path = Path("archive/review-results/model-credit-risk-v12.json")
result_path.parent.mkdir(parents=True, exist_ok=True)
result_path.write_bytes(canonical(evidence))
print(result_path)

set -euo pipefail
python archive/verify_hybrid_statement.py
cosign sign-blob --yes --key keys/archive-review.key --bundle archive/review-results/model-credit-risk-v12.cosign.sig archive/review-results/model-credit-risk-v12.json

Action: Store the artifact, provenance, canonical statement, digest-named chain nodes, both statement signatures, signed verifier policy, protected policy digest, and signed verification result in WORM storage. A missing, expired, replayed, or unverifiable proof is a hard restore failure.

Verify safely

A read-only identity that cannot alter the controlled asset must obtain canonical statement bytes, independent key registries, detached signatures, cryptographic policy, and prior continuity receipts; validate the effective asset, policy, input, and scope versions or digests; then re-canonicalize and hash the statement, independently verify both signature families and expiry, and traverse every signed rollover link without treating one signature family as sufficient.

AID-M-002.005-G002

Concept:

Policy renewal and statement rollover are separate operations. Renew the 30-day signed verifier policy without changing the approved head when only the approval window changes. Create a new statement only before its 180-day expiry, when a trust anchor or accepted profile changes, or after an approved archive migration. Every new statement must name the prior head, while every new policy pins the new head and the original chain root.

Rollover the statement without losing prior proofs

set -euo pipefail
export APPROVED_ARCHIVE_POLICY_SHA256
python archive/verify_hybrid_statement.py

OLD_HEAD="$(sha256sum archive/model-credit-risk-v12.provenance.json | cut -d' ' -f1)"
CHAIN_ROOT="$(python -c 'import json; print(json.load(open("archive/verifier_policy.json"))["assets"]["model-credit-risk-v12"]["chain_root_sha256"])')"
mkdir -p archive/statements archive/policies
cp --no-clobber archive/model-credit-risk-v12.cosign.sig "archive/statements/${OLD_HEAD}.cosign.sig"
cp --no-clobber archive/model-credit-risk-v12.mldsa.sig "archive/statements/${OLD_HEAD}.mldsa.sig"
cp --no-clobber archive/verifier_policy.json "archive/policies/${OLD_HEAD}.policy.json"
cp --no-clobber archive/verifier_policy.cosign.sig "archive/policies/${OLD_HEAD}.policy.cosign.sig"

KMS_KEY_ARN="$(aws kms describe-key --key-id alias/aidefend-archive-mldsa --query KeyMetadata.Arn --output text)"
test -n "$KMS_KEY_ARN"
export ML_DSA_KMS_KEY_ARN="$KMS_KEY_ARN"
export PREVIOUS_STATEMENT_SHA256="$OLD_HEAD"
export CHAIN_ROOT_SHA256="$CHAIN_ROOT"
python archive/create_hybrid_statement.py

cosign sign-blob --yes --key keys/archive-classical.key --bundle archive/model-credit-risk-v12.cosign.sig archive/model-credit-risk-v12.provenance.json
aws kms sign --key-id "$KMS_KEY_ARN" --message fileb://archive/model-credit-risk-v12.provenance.json --message-type RAW --signing-algorithm ML_DSA_SHAKE_256 --query Signature --output text | base64 --decode > archive/model-credit-risk-v12.mldsa.sig

export ARCHIVE_POLICY_VERSION="$(date -u +%Y%m%d%H%M%S)"
python archive/build_verifier_policy.py
cosign sign-blob --yes --key keys/verifier-policy-root.key --bundle archive/verifier_policy.cosign.sig archive/verifier_policy.json
NEW_POLICY_SHA256="$(sha256sum archive/verifier_policy.json | cut -d' ' -f1)"
printf 'submit protected APPROVED_ARCHIVE_POLICY_SHA256=%s\n' "$NEW_POLICY_SHA256"
# An independent approval job publishes that digest. The restore job never derives it from this archive.

Independent activation gate

set -euo pipefail
test "$APPROVED_ARCHIVE_POLICY_SHA256" = "$(sha256sum archive/verifier_policy.json | cut -d' ' -f1)"
python archive/verify_hybrid_statement.py
NEW_HEAD="$(sha256sum archive/model-credit-risk-v12.provenance.json | cut -d' ' -f1)"
cp --no-clobber archive/model-credit-risk-v12.cosign.sig "archive/statements/${NEW_HEAD}.cosign.sig"
cp --no-clobber archive/model-credit-risk-v12.mldsa.sig "archive/statements/${NEW_HEAD}.mldsa.sig"
# Replicate the digest-named statement, signatures, policies, and signed review result to WORM storage.

Freshness: Run the verifier daily and before every restore. Renew the signed verifier policy before its 30-day expiry, and roll the statement no later than 30 days before its 180-day expiry. The approval job must reject a head whose previous digest is not the currently approved head or whose chain does not terminate at the signed root.

Action: Emergency restore pressure does not authorize bypassing the protected policy digest, either signature, provenance digest, current-head approval, or chain verification.

Open-source tools

• OpenSSL 3.5+
• liboqs (prototype ML-DSA interoperability testing only; not recommended for production)
• Cosign
• in-toto
• slsa-verifier
• python-tuf

Source-available / open-weight tools

• immudb (Business Source License 1.1; source-available)

Commercial / hosted tools

• AWS Key Management Service (ML-DSA signing keys)
• DigiCert ONE
• Keyfactor Command
• Amazon S3 Object Lock
• Amazon S3 Glacier Vault Lock
• Azure Immutable Blob Storage

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise
• AML.T0115 Publish Poisoned AI Artifacts
• AML.T0010.002 AI Supply Chain Compromise: Data
• AML.T0010.003 AI Supply Chain Compromise: Model
• AML.T0010.004 AI Supply Chain Compromise: Container Registry
• AML.T0115.001 Publish Poisoned AI Artifacts: Models (detects post-approval artifact substitution; not poisoning during training)
• AML.T0059 Erode Dataset Integrity
• AML.T0076 Corrupt AI Model
• AML.T0112.001 Machine Compromise: AI Artifacts

MAESTRO

• Data Tampering (L2)
• Supply Chain Attacks (Cross-Layer)
• Compromised Container Images (L4)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain
• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks
• ML10:2023 Model Poisoning (detects post-approval model substitution; not poisoning during training)

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities

NIST Adversarial Machine Learning 2025

• NISTAML.051 Model Poisoning (Supply Chain)

Cisco Integrated AI Security and Safety Framework

• AITech-6.1 Training Data Poisoning (Verifies long-lived dataset bytes before their reuse after archive or trust migration)

Google Secure AI Framework 2.0 - Risks

• MST: Model Source Tampering
• MDT: Model Deployment Tampering

Databricks AI Security Framework 3.0

• Raw Data 1.6: Insufficient data lineage
• Raw Data 1.7: Lack of data trustworthiness
• Governance 4.1: Lack of traceability and transparency of model assets
• Model 7.3: ML Supply chain vulnerabilities