AID-R-001.001: Versioned Model Rollback & Restoration

Version 1.20261015 · Actionable sub-technique

Restore an incident-affected model-serving release by deploying an exact known-good version from a retention-enforced immutable backup, managed-model deployment record, prompt or guardrail configuration manifest, or adapter attachment manifest authorized by an authority-signed recovery manifest, then verify the live release through an identity outside the backup-writer and deployment trust domains.

Pillars: model

Lifecycle: improvement

Scope boundary

This sub-technique owns restoration of an incident-affected model-serving release by selecting and deploying an exact authority-approved pre-incident model and serving-image pair, hosted deployment record, prompt or guardrail configuration, or adapter attachment set from immutable versioned state, then reading the live identity back through an identity outside the backup-writer and deployment trust domains before cutover. It applies only when a trustworthy prior version and trusted cutoff exist. Registry labels, endpoint names, hosted model IDs, and tool success do not authorize recovery or establish live identity. It does not quarantine a suspect model, prove adapter eviction, or build a newly remediated model.

AID-R-001.002: AID-R-001.001 restores an exact trusted pre-incident version; AID-R-001.002 creates a newly remediated model through clean retraining or recovery learning. Use rollback when immutable trusted prior state exists. Use AID-R-001.002 when rollback is unavailable, incomplete, or untrustworthy; AID-R-001.001 must not silently fall back to retraining.

AID-E-003.001: AID-E-003.001 quarantines the suspect model; AID-R-001.001 deploys and verifies a separate known-good version. AID-E-003.001 makes the suspect digest ineligible for serving, preserves it as incident evidence, and routes the incident to recovery. Successful rollback does not prove that the compromised model was quarantined everywhere.

AID-E-003.005: AID-E-003.005 evicts compromised adapters and derivatives; AID-R-001.001 restores and verifies known-good model or adapter state. AID-R-001.001 consumes quarantine and tombstone evidence but owns only the rollback and live-release verification. A restored deployment does not prove incident-time adapter eviction across serving, registries, caches, and lineage.

Implementation guidance

AID-R-001.001-G001

Maintain Object-Locked model versions and authority-signed digest-bound recovery manifests.
Concept:

Versioning alone permits privileged deletion and is not immutable retention. Create the bucket with S3 Object Lock enabled, enable versioning, enforce a retention mode and KMS encryption, and keep runtime and deployment identities outside the backup-writer role. Retention duration is a reviewed system policy input, not a framework-wide constant.

Provision retention-enforced model backups
# File: infrastructure/model_backups.tf
variable "model_backup_bucket_name" {
  type = string
}

variable "model_backup_retention_days" {
  type = number
  validation {
    condition     = var.model_backup_retention_days > 0
    error_message = "Retention must be a positive approved policy value."
  }
}

variable "model_backup_kms_key_arn" {
  type = string
}

resource "aws_s3_bucket" "model_backups" {
  bucket              = var.model_backup_bucket_name
  object_lock_enabled = true
}

resource "aws_s3_bucket_versioning" "model_backups" {
  bucket = aws_s3_bucket.model_backups.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_object_lock_configuration" "model_backups" {
  bucket = aws_s3_bucket.model_backups.id
  rule {
    default_retention {
      mode = "COMPLIANCE"
      days = var.model_backup_retention_days
    }
  }
  depends_on = [aws_s3_bucket_versioning.model_backups]
}

resource "aws_s3_bucket_server_side_encryption_configuration" "model_backups" {
  bucket = aws_s3_bucket.model_backups.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = var.model_backup_kms_key_arn
    }
    bucket_key_enabled = true
  }
}

resource "aws_s3_bucket_public_access_block" "model_backups" {
  bucket                  = aws_s3_bucket.model_backups.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}
Independent promotion evidence

For every promoted model, the backup writer records the bucket, key, immutable version ID, object digest, serving-image digest, build attestation digest, and trusted cutoff. A separate release-security identity verifies the uploaded version by reading it back, recomputing its digest, and signs the canonical recovery manifest with a key unavailable to backup writers, runtime services, and deployers. Retain that manifest and signature in a separately administered Object-Locked evidence bucket or transparency log.

Verify the result

Action: Test retention with denied mutations and independent periodic readback for the complete signed population; never treat upload, registry tag, or version ID alone as restore-capable evidence.

AID-R-001.001-G002

Restore an authority-selected immutable model version and read the live model and serving-image identity back through a separate deployment read path.
When provider recovery applies:

Use rollback only when the recovery authority has selected a pre-incident model and serving-image pair from a trusted cutoff. A model registry may locate candidates, but registry tags do not authorize recovery. The signed manifest is the trust root and must identify the exact immutable object version and digests.

Verify the authority-signed manifest and immutable version
{
  "schema_version": "aidefend.r001001-recovery-manifest/v1",
  "control": "AID-R-001.001",
  "system_or_tenant": "example-system",
  "incident_id": "INC-EXAMPLE",
  "release_id": "recovery-example",
  "trusted_cutoff_at": "2026-07-10T10:00:00Z",
  "model": {
    "bucket": "model-backups",
    "key": "models/example/model.tar.zst",
    "version_id": "IMMUTABLE_VERSION_ID",
    "sha256": "2e1301125954d3ee79208b29e1a1eff74449619f20414dc7eef4bf744b6ddf52"
  },
  "serving_image_digest": "registry.example.invalid/inference@sha256:c3ef7782ff3a7ea742414d953744f4b7491fdf99fbea3a61ae836a915726ee45",
  "runtime": {
    "connect_timeout_seconds": 5,
    "read_timeout_seconds": 60,
    "maximum_retry_attempts": 3,
    "maximum_manifest_bytes": 262144,
    "maximum_model_bytes": 8589934592,
    "signature_timeout_seconds": 30
  }
}
: "${AIDEFEND_BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES:?load from the versioned recovery runtime profile}"
: "${AIDEFEND_BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS:?load from the same versioned recovery runtime profile}"
python restore/fetch_verified_model.py \
  --manifest recovery-manifest.json \
  --manifest-bundle recovery-manifest.json.sigstore.json \
  --authority-key keys/recovery-authority.pub \
  --output recovered-model.tar.zst

# File: restore/fetch_verified_model.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import boto3
from botocore.config import Config

SHA256 = re.compile(r"^[0-9a-f]{64}$")
IMAGE = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$")
BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES = int(
    os.environ["AIDEFEND_BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES"]
)
BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(
    os.environ["AIDEFEND_BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS"]
)


def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for name, item in pairs:
        if name in value:
            raise ValueError(f"duplicate JSON key: {name}")
        value[name] = item
    return value


def strict_json(raw: bytes) -> dict[str, Any]:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError("manifest must contain one JSON object")
    return value


def positive(value: object, name: str, *, integer: bool = False) -> float | int:
    if (
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not math.isfinite(value)
        or value <= 0
        or (integer and not isinstance(value, int))
    ):
        raise ValueError(f"{name} must be finite and positive")
    return value


def read_limited(path: Path, maximum: int) -> bytes:
    with path.open("rb") as source:
        raw = source.read(maximum + 1)
    if len(raw) > maximum:
        raise ValueError(f"{path} exceeds the configured bound")
    return raw


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


def verified_manifest(
    path: Path,
    bundle: Path,
    public_key: str,
) -> tuple[dict[str, Any], bytes]:
    if BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES <= 0:
        raise RuntimeError("bootstrap trust-document bound is invalid")
    signature_timeout = float(
        positive(
            BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS,
            "bootstrap signature verification timeout",
        )
    )
    raw = read_limited(path, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES)
    bundle_raw = read_limited(bundle, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES)
    if not raw or not bundle_raw:
        raise ValueError("manifest and Sigstore bundle must both be non-empty")
    with tempfile.TemporaryDirectory(prefix="r001001-provider-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        manifest_snapshot = snapshot_root / "recovery-manifest.json"
        bundle_snapshot = snapshot_root / "recovery-manifest.json.sigstore.json"
        write_private_snapshot(manifest_snapshot, raw)
        write_private_snapshot(bundle_snapshot, bundle_raw)
        subprocess.run(
            [
                "cosign",
                "verify-blob",
                "--key",
                public_key,
                "--bundle",
                str(bundle_snapshot),
                str(manifest_snapshot),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=signature_timeout,
        )
        if read_limited(manifest_snapshot, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES) != raw:
            raise RuntimeError("verified manifest snapshot changed")
        if read_limited(bundle_snapshot, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES) != bundle_raw:
            raise RuntimeError("verified Sigstore bundle snapshot changed")
    document = strict_json(raw)
    runtime = document.get("runtime")
    if not isinstance(runtime, dict):
        raise ValueError("signed runtime is required")
    positive(runtime.get("signature_timeout_seconds"), "signed signature timeout")
    maximum_manifest = int(
        positive(runtime.get("maximum_manifest_bytes"), "maximum manifest bytes", integer=True)
    )
    if len(raw) > maximum_manifest:
        raise ValueError("manifest exceeds its signed bound")
    return document, raw


def required_text(value: object, name: str) -> str:
    if not isinstance(value, str) or not value or "\x00" in value:
        raise ValueError(f"{name} must be a non-empty string")
    return value


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--manifest-bundle", type=Path, required=True)
    parser.add_argument("--authority-key", required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    if args.manifest_bundle != Path(str(args.manifest) + ".sigstore.json"):
        raise ValueError("manifest bundle must be the sibling <manifest>.sigstore.json")

    manifest, raw_manifest = verified_manifest(
        args.manifest, args.manifest_bundle, args.authority_key
    )
    root_fields = {
        "schema_version",
        "control",
        "system_or_tenant",
        "incident_id",
        "release_id",
        "trusted_cutoff_at",
        "model",
        "serving_image_digest",
        "runtime",
    }
    if (
        set(manifest) != root_fields
        or manifest.get("schema_version")
        != "aidefend.r001001-recovery-manifest/v1"
        or manifest.get("control") != "AID-R-001.001"
    ):
        raise ValueError("unexpected recovery-manifest schema")
    for name in ("system_or_tenant", "incident_id", "release_id"):
        required_text(manifest.get(name), name)
    cutoff_text = required_text(
        manifest.get("trusted_cutoff_at"), "trusted_cutoff_at"
    )
    if not cutoff_text.endswith("Z"):
        raise ValueError("trusted_cutoff_at must be an RFC 3339 UTC timestamp")
    try:
        cutoff = datetime.fromisoformat(cutoff_text[:-1] + "+00:00")
    except ValueError as exc:
        raise ValueError("trusted_cutoff_at is malformed") from exc
    if cutoff.tzinfo is None or cutoff.utcoffset() is None:
        raise ValueError("trusted_cutoff_at must include a timezone")
    cutoff = cutoff.astimezone(timezone.utc)
    if not isinstance(manifest.get("serving_image_digest"), str) or IMAGE.fullmatch(
        manifest["serving_image_digest"]
    ) is None:
        raise ValueError("serving image must be pinned by SHA-256 digest")

    model = manifest.get("model")
    model_fields = {"bucket", "key", "version_id", "sha256"}
    if not isinstance(model, dict) or set(model) != model_fields:
        raise ValueError("model locator schema differs")
    for name in ("bucket", "key", "version_id"):
        required_text(model.get(name), f"model.{name}")
    if (
        not isinstance(model.get("sha256"), str)
        or SHA256.fullmatch(model["sha256"]) is None
        or set(model["sha256"]) == {"0"}
    ):
        raise ValueError("model.sha256 must be a nonzero lowercase SHA-256 digest")

    runtime = manifest["runtime"]
    runtime_fields = {
        "connect_timeout_seconds",
        "read_timeout_seconds",
        "maximum_retry_attempts",
        "maximum_manifest_bytes",
        "maximum_model_bytes",
        "signature_timeout_seconds",
    }
    if not isinstance(runtime, dict) or set(runtime) != runtime_fields:
        raise ValueError("signed runtime schema differs")
    connect_timeout = float(
        positive(runtime["connect_timeout_seconds"], "connect timeout")
    )
    read_timeout = float(positive(runtime["read_timeout_seconds"], "read timeout"))
    maximum_retry_attempts = int(
        positive(
            runtime["maximum_retry_attempts"],
            "maximum retry attempts",
            integer=True,
        )
    )
    maximum_model = int(
        positive(runtime["maximum_model_bytes"], "maximum model bytes", integer=True)
    )

    client = boto3.client(
        "s3",
        config=Config(
            connect_timeout=connect_timeout,
            read_timeout=read_timeout,
            retries={"mode": "standard", "max_attempts": maximum_retry_attempts},
        ),
    )
    response = client.get_object(
        Bucket=model["bucket"],
        Key=model["key"],
        VersionId=model["version_id"],
        ChecksumMode="ENABLED",
    )
    if (
        response.get("VersionId") != model["version_id"]
        or response.get("ResponseMetadata", {}).get("HTTPStatusCode") != 200
    ):
        response["Body"].close()
        raise RuntimeError("S3 readback did not return the exact immutable version")
    last_modified = response.get("LastModified")
    if (
        not isinstance(last_modified, datetime)
        or last_modified.tzinfo is None
        or last_modified.utcoffset() is None
    ):
        response["Body"].close()
        raise RuntimeError("S3 LastModified is absent or malformed")
    last_modified = last_modified.astimezone(timezone.utc)
    if last_modified > cutoff:
        response["Body"].close()
        raise RuntimeError("S3 immutable version was modified after the trusted cutoff")
    content_length = response.get("ContentLength")
    if (
        isinstance(content_length, bool)
        or not isinstance(content_length, int)
        or content_length <= 0
        or content_length > maximum_model
    ):
        response["Body"].close()
        raise RuntimeError("S3 object length is absent, empty, or outside the signed bound")

    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    descriptor = os.open(args.output, flags, 0o600)
    digest = hashlib.sha256()
    observed = 0
    try:
        with os.fdopen(descriptor, "wb") as handle:
            try:
                for chunk in iter(lambda: response["Body"].read(1024 * 1024), b""):
                    observed += len(chunk)
                    if observed > maximum_model:
                        raise RuntimeError("model stream exceeds the signed byte bound")
                    digest.update(chunk)
                    handle.write(chunk)
            finally:
                response["Body"].close()
            handle.flush()
            os.fsync(handle.fileno())
        if observed != content_length or digest.hexdigest() != model["sha256"]:
            raise RuntimeError("restored model length or digest differs")
    except BaseException:
        args.output.unlink(missing_ok=True)
        raise

    print(
        json.dumps(
            {
                "schema_version": "aidefend.r001001-provider-fetch/v1",
                "control": "AID-R-001.001",
                "release_id": manifest["release_id"],
                "manifest_sha256": hashlib.sha256(raw_manifest).hexdigest(),
                "model_sha256": digest.hexdigest(),
                "model_bytes": observed,
                "s3_version_id": response["VersionId"],
                "s3_last_modified": last_modified.isoformat().replace("+00:00", "Z"),
                "trusted_cutoff_at": cutoff.isoformat().replace("+00:00", "Z"),
                "serving_image_digest": manifest["serving_image_digest"],
                "result": "PASS",
            },
            sort_keys=True,
            separators=(",", ":"),
        )
    )


if __name__ == "__main__":
    main()
Deploy and verify outside the deployment trust domain

Deploy the verified bytes and the exact image digest through the approved rollback workflow; never substitute a model name, mutable tag, default registry version, or serialization-specific filename. The deployment owner signs a receipt binding the authority manifest, fetched model digest, image digest, workflow run, and target release. A separate read-only verifier fetches workload identity and readiness data directly from the orchestrator and service endpoint, verifies that the running model digest and image digest exactly match the manifest, replays the approved smoke suite, and signs aidefend.r001001-verification/v1.

Fail semantics: Missing manifest bundle, inaccessible version, retention failure, digest mismatch, mutable image tag, post-cutoff artifact, deployment/verification key reuse, stale verification, or inability to observe the live model and image identity blocks cutover. Tool success or a healthy endpoint without exact identity is not proof.

AID-R-001.001-G003

Restore hosted fine-tuned models, prompt/guardrail configurations, and LoRA or adapter attachments from a signed provider recovery manifest.
Use a provider-bound, signed recovery snapshot

For Azure AI Services, bind the exact subscription, resource group, account, deployment, model, SKU, RAI policy, managed identity, API version, immutable prompt/adapter digests, and runtime limits in one recovery-authority-signed manifest. The example below uses the documented 2024-10-01 deployment API. Other providers need their own response schema; an Azure control-plane readback must never be used as proof of application-owned prompt or adapter bytes.

{
  "schema_version": "aidefend.r001001-hosted-model-manifest/v1",
  "control": "AID-R-001.001",
  "release_id": "recovery-20260720-01",
  "incident_id": "INC-20260720-001",
  "management_origin": "https://management.azure.com",
  "api_version": "2024-10-01",
  "deployment": {
    "subscription_id": "00000000-0000-4000-8000-000000000001",
    "resource_group": "rg-ai-production",
    "account_name": "aoai-production",
    "deployment_name": "support-bot-production",
    "managed_identity_client_id": "11111111-1111-4111-8111-111111111111",
    "model_format": "OpenAI",
    "model_name": "gpt-4.1-mini-2025-04-14.ft-production",
    "model_version": "1",
    "sku_name": "Standard",
    "capacity": 50,
    "rai_policy_name": "production-low-risk-v7"
  },
  "prompt_bundle": {"artifact_uri": "s3://ai-release-bom/support-bot/prompt-v41.json", "sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
  "adapter_bundle": {"enabled": false, "artifact_uri": null, "sha256": null},
  "application": {
    "origin": "https://release-state.example.invalid",
    "target_id": "support-bot-production",
    "release_id": "recovery-20260720-01",
    "consumer_ids": ["support-bot-replica-a", "support-bot-replica-b"]
  },
  "runtime": {"connect_timeout_seconds": 5, "read_timeout_seconds": 20, "operation_timeout_seconds": 600, "poll_interval_seconds": 5, "signature_timeout_seconds": 30, "maximum_manifest_bytes": 262144, "maximum_response_bytes": 1048576, "maximum_artifact_bytes": 16777216}
}

Before reading this untrusted manifest, load AIDEFEND_BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES and AIDEFEND_BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS from the same local, versioned recovery runtime profile. Manifest runtime values become usable only after the manifest and its Sigstore bundle verify as an unchanged pair.

Verify the immutable snapshot, restore with an in-process identity, and bound every read
# File: restore/restore_hosted_azure.py
from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any
from urllib.parse import quote, urlsplit

import httpx
import requests as azure_requests
from azure.core.pipeline.transport import RequestsTransport
from azure.identity import ManagedIdentityCredential

BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES = int(os.environ["AIDEFEND_BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES"])
BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["AIDEFEND_BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
HEX = set("0123456789abcdef")
ROOT_FIELDS = {"schema_version", "control", "release_id", "incident_id", "management_origin", "api_version", "deployment", "prompt_bundle", "adapter_bundle", "application", "runtime"}
APPLICATION_FIELDS = {"origin", "target_id", "release_id", "consumer_ids"}
DEPLOYMENT_FIELDS = {"subscription_id", "resource_group", "account_name", "deployment_name", "managed_identity_client_id", "model_format", "model_name", "model_version", "sku_name", "capacity", "rai_policy_name"}
RUNTIME_FIELDS = {"connect_timeout_seconds", "read_timeout_seconds", "operation_timeout_seconds", "poll_interval_seconds", "signature_timeout_seconds", "maximum_manifest_bytes", "maximum_response_bytes", "maximum_artifact_bytes"}


def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for name, item in pairs:
        if name in value:
            raise ValueError(f"duplicate JSON key: {name}")
        value[name] = item
    return value


def strict_json(raw: bytes) -> dict[str, Any]:
    value = json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique_object, parse_constant=reject_constant)
    if not isinstance(value, dict):
        raise ValueError("one JSON object is required")
    return value


def canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")


def digest(value: object, name: str) -> str:
    if not isinstance(value, str) or len(value) != 64 or set(value) - HEX or set(value) == {"0"}:
        raise ValueError(f"{name} must be a nonzero lowercase SHA-256 digest")
    return value


def positive(value: object, name: str, *, integer: bool = False) -> float | int:
    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
        raise ValueError(f"{name} must be finite and positive")
    if integer and not isinstance(value, int):
        raise ValueError(f"{name} must be an integer")
    return value


def read_limited(path: Path, maximum: int) -> bytes:
    with path.open("rb") as source:
        raw = source.read(maximum + 1)
    if len(raw) > maximum:
        raise ValueError(f"{path} exceeds the signed size bound")
    return raw


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


def verified_snapshot(path: Path, bundle: Path, public_key: str) -> tuple[dict[str, Any], bytes]:
    if BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES < 1:
        raise RuntimeError("verified bootstrap profile has an invalid trust-document bound")
    deadline = float(positive(BOOTSTRAP_SIGNATURE_VERIFY_TIMEOUT_SECONDS, "bootstrap signature verification timeout"))
    raw = read_limited(path, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES)
    bundle_raw = read_limited(bundle, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES)
    if not raw or not bundle_raw:
        raise ValueError("manifest and Sigstore bundle must both be non-empty")
    with tempfile.TemporaryDirectory(prefix="r001001-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        manifest_snapshot = snapshot_root / "hosted-model-manifest.json"
        bundle_snapshot = snapshot_root / "hosted-model-manifest.json.sigstore.json"
        write_private_snapshot(manifest_snapshot, raw)
        write_private_snapshot(bundle_snapshot, bundle_raw)
        subprocess.run(["cosign", "verify-blob", "--key", public_key, "--bundle", str(bundle_snapshot), str(manifest_snapshot)], check=True, capture_output=True, text=True, timeout=deadline)
        if read_limited(manifest_snapshot, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES) != raw:
            raise RuntimeError("verified manifest snapshot changed")
        if read_limited(bundle_snapshot, BOOTSTRAP_MAXIMUM_TRUST_DOCUMENT_BYTES) != bundle_raw:
            raise RuntimeError("verified Sigstore bundle snapshot changed")
    document = strict_json(raw)
    runtime = document.get("runtime")
    if not isinstance(runtime, dict) or set(runtime) != RUNTIME_FIELDS:
        raise ValueError("signed runtime schema differs")
    positive(runtime["signature_timeout_seconds"], "signed signature timeout")
    if len(raw) > positive(runtime["maximum_manifest_bytes"], "maximum manifest bytes", integer=True):
        raise ValueError("manifest exceeds its signed size bound")
    return document, raw


def read_response(response: httpx.Response, maximum: int) -> dict[str, Any]:
    if 300 <= response.status_code < 400 or response.headers.get("location"):
        raise RuntimeError("redirects are forbidden")
    if response.headers.get("content-encoding", "identity").lower() != "identity":
        raise RuntimeError("unexpected content encoding")
    media_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
    if media_type != "application/json":
        raise RuntimeError("Azure response is not application/json")
    chunks: list[bytes] = []
    observed = 0
    for chunk in response.iter_bytes():
        observed += len(chunk)
        if observed > maximum:
            raise RuntimeError("Azure response exceeds the signed decoded-body bound")
        chunks.append(chunk)
    response.raise_for_status()
    return strict_json(b"".join(chunks))


def request_json(client: httpx.Client, method: str, url: str, maximum: int, *, body: dict[str, Any] | None = None, extra_headers: dict[str, str] | None = None) -> dict[str, Any]:
    headers = {"Accept": "application/json", "Accept-Encoding": "identity"}
    content = None
    if body is not None:
        headers["Content-Type"] = "application/json"
        content = canonical(body)
    if extra_headers is not None:
        headers.update(extra_headers)
    with client.stream(method, url, headers=headers, content=content) as response:
        return read_response(response, maximum)


def verify_artifact(spec: dict[str, Any], path: Path | None, maximum: int, label: str) -> str | None:
    if label == "adapter" and spec == {"enabled": False, "artifact_uri": None, "sha256": None}:
        if path is not None:
            raise ValueError("disabled adapter must not receive bytes")
        return None
    required = {"artifact_uri", "sha256"} if label == "prompt" else {"enabled", "artifact_uri", "sha256"}
    if set(spec) != required or (label == "adapter" and spec.get("enabled") is not True) or not isinstance(spec.get("artifact_uri"), str) or path is None:
        raise ValueError(f"{label} artifact schema differs")
    expected = digest(spec.get("sha256"), f"{label} sha256")
    raw = read_limited(path, maximum)
    if not hashlib.sha256(raw).hexdigest() == expected:
        raise RuntimeError(f"{label} artifact digest differs")
    return expected


def exact_origin(value: object) -> str:
    if not isinstance(value, str):
        raise ValueError("application origin must be a string")
    parsed = urlsplit(value)
    if parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.path not in {"", "/"} or parsed.query or parsed.fragment:
        raise ValueError("application origin must be one exact HTTPS origin")
    return f"https://{parsed.netloc.lower()}"


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--manifest", type=Path, required=True)
    parser.add_argument("--manifest-bundle", type=Path, required=True)
    parser.add_argument("--authority-key", required=True)
    parser.add_argument("--prompt-bundle", type=Path, required=True)
    parser.add_argument("--adapter-bundle", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    if args.manifest_bundle != Path(str(args.manifest) + ".sigstore.json"):
        raise ValueError("manifest bundle must be the sibling <manifest>.sigstore.json")

    manifest, raw_manifest = verified_snapshot(args.manifest, args.manifest_bundle, args.authority_key)
    if set(manifest) != ROOT_FIELDS or manifest.get("schema_version") != "aidefend.r001001-hosted-model-manifest/v1" or manifest.get("control") != "AID-R-001.001":
        raise ValueError("hosted recovery manifest schema differs")
    if manifest["management_origin"] != "https://management.azure.com" or manifest["api_version"] != "2024-10-01":
        raise ValueError("unexpected Azure management origin or API version")
    deployment = manifest["deployment"]
    runtime = manifest["runtime"]
    if not isinstance(deployment, dict) or set(deployment) != DEPLOYMENT_FIELDS:
        raise ValueError("deployment schema differs")
    for name in DEPLOYMENT_FIELDS - {"capacity", "managed_identity_client_id"}:
        if not isinstance(deployment[name], str) or not deployment[name]:
            raise ValueError(f"deployment {name} is required")
    if deployment["managed_identity_client_id"] is not None and (not isinstance(deployment["managed_identity_client_id"], str) or not deployment["managed_identity_client_id"]):
        raise ValueError("managed identity client ID is invalid")
    positive(deployment["capacity"], "capacity", integer=True)
    maximum_response = int(positive(runtime["maximum_response_bytes"], "maximum response bytes", integer=True))
    maximum_artifact = int(positive(runtime["maximum_artifact_bytes"], "maximum artifact bytes", integer=True))
    prompt_digest = verify_artifact(manifest["prompt_bundle"], args.prompt_bundle, maximum_artifact, "prompt")
    adapter_digest = verify_artifact(manifest["adapter_bundle"], args.adapter_bundle, maximum_artifact, "adapter")
    application = manifest["application"]
    if not isinstance(application, dict) or set(application) != APPLICATION_FIELDS:
        raise ValueError("application restore schema differs")
    application_origin = exact_origin(application["origin"])
    if application["release_id"] != manifest["release_id"]:
        raise RuntimeError("application release binding differs")
    for name in ("target_id", "release_id"):
        if not isinstance(application[name], str) or not application[name] or any(character in application[name] for character in "/\\?#\x00"):
            raise ValueError(f"unsafe application {name}")
    consumers = application["consumer_ids"]
    if not isinstance(consumers, list) or not consumers or consumers != sorted(set(consumers)) or any(not isinstance(item, str) or not item or any(character in item for character in "/\\?#\x00") for item in consumers):
        raise ValueError("application consumer population must be sorted, unique, and non-empty")

    for name in ("subscription_id", "resource_group", "account_name", "deployment_name"):
        if any(character in deployment[name] for character in "/\\?#\x00"):
            raise ValueError(f"deployment {name} contains a path delimiter")
    segments = ["subscriptions", deployment["subscription_id"], "resourceGroups", deployment["resource_group"], "providers", "Microsoft.CognitiveServices", "accounts", deployment["account_name"], "deployments", deployment["deployment_name"]]
    resource_path = "/" + "/".join(str(segment) for segment in segments)
    request_path = "/" + "/".join(quote(str(segment), safe="") for segment in segments)
    url = manifest["management_origin"] + request_path + "?api-version=" + quote(manifest["api_version"], safe="")
    expected_id = resource_path.lower()
    identity_session = azure_requests.Session()
    identity_session.trust_env = False
    identity_transport = RequestsTransport(session=identity_session, session_owner=True, use_env_settings=False, connection_timeout=float(runtime["connect_timeout_seconds"]), read_timeout=float(runtime["read_timeout_seconds"]))
    credential = ManagedIdentityCredential(client_id=deployment["managed_identity_client_id"], transport=identity_transport)
    try:
        token = credential.get_token("https://management.azure.com/.default").token
    finally:
        credential.close()
    timeout = httpx.Timeout(connect=float(positive(runtime["connect_timeout_seconds"], "connect timeout")), read=float(positive(runtime["read_timeout_seconds"], "read timeout")), write=float(runtime["read_timeout_seconds"]), pool=float(runtime["connect_timeout_seconds"]))
    body = {"sku": {"name": deployment["sku_name"], "capacity": deployment["capacity"]}, "properties": {"model": {"format": deployment["model_format"], "name": deployment["model_name"], "version": deployment["model_version"]}, "raiPolicyName": deployment["rai_policy_name"]}}
    with httpx.Client(headers={"Authorization": "Bearer " + token}, timeout=timeout, follow_redirects=False, trust_env=False) as client:
        request_json(client, "PUT", url, maximum_response, body=body)
        deadline = time.monotonic() + float(positive(runtime["operation_timeout_seconds"], "operation timeout"))
        observed: dict[str, Any] = {}
        while time.monotonic() < deadline:
            observed = request_json(client, "GET", url, maximum_response)
            state = (observed.get("properties") or {}).get("provisioningState")
            if state == "Succeeded":
                break
            if state in {"Failed", "Canceled"}:
                raise RuntimeError(f"Azure deployment ended in {state}")
            time.sleep(float(positive(runtime["poll_interval_seconds"], "poll interval")))
        else:
            raise TimeoutError("Azure deployment did not converge before the signed deadline")

    properties = observed.get("properties")
    model = properties.get("model") if isinstance(properties, dict) else None
    sku = observed.get("sku")
    if not isinstance(properties, dict) or not isinstance(model, dict) or not isinstance(sku, dict):
        raise RuntimeError("Azure deployment readback schema differs")
    expected = {"id": expected_id, "name": deployment["deployment_name"], "model_format": deployment["model_format"], "model_name": deployment["model_name"], "model_version": deployment["model_version"], "sku_name": deployment["sku_name"], "capacity": deployment["capacity"], "rai_policy_name": deployment["rai_policy_name"]}
    actual = {"id": str(observed.get("id", "")).rstrip("/").lower(), "name": observed.get("name"), "model_format": model.get("format"), "model_name": model.get("name"), "model_version": model.get("version"), "sku_name": sku.get("name"), "capacity": sku.get("capacity"), "rai_policy_name": properties.get("raiPolicyName")}
    if actual != expected:
        raise RuntimeError(f"Azure deployment readback differs: {actual}")

    application_body = {
        "release_id": application["release_id"],
        "target_id": application["target_id"],
        "prompt_bundle": manifest["prompt_bundle"],
        "adapter_bundle": manifest["adapter_bundle"],
        "expected_consumer_ids": consumers,
    }
    application_path = "/v1/releases/" + quote(application["release_id"], safe="") + "/targets/" + quote(application["target_id"], safe="") + "/effective-state"
    idempotency_key = manifest["release_id"] + ":" + hashlib.sha256(canonical(application_body)).hexdigest()
    application_timeout = httpx.Timeout(connect=float(positive(runtime["connect_timeout_seconds"], "connect timeout")), read=float(positive(runtime["read_timeout_seconds"], "read timeout")), write=float(runtime["read_timeout_seconds"]), pool=float(runtime["connect_timeout_seconds"]))
    expected_consumers = [
        {
            "consumer_id": consumer_id,
            "release_id": application["release_id"],
            "prompt_sha256": prompt_digest,
            "adapter_sha256": adapter_digest,
            "loaded": True,
        }
        for consumer_id in consumers
    ]
    with httpx.Client(
        verify=os.environ["APPLICATION_CA_FILE"],
        cert=(os.environ["APPLICATION_CLIENT_CERT"], os.environ["APPLICATION_CLIENT_KEY"]),
        timeout=application_timeout,
        follow_redirects=False,
        trust_env=False,
    ) as application_client:
        request_json(
            application_client,
            "PUT",
            application_origin + application_path,
            maximum_response,
            body=application_body,
            extra_headers={"Idempotency-Key": idempotency_key},
        )
        deadline = time.monotonic() + float(positive(runtime["operation_timeout_seconds"], "operation timeout"))
        application_observed: dict[str, Any] = {}
        while time.monotonic() < deadline:
            application_observed = request_json(application_client, "GET", application_origin + application_path, maximum_response)
            exact_application_state = {
                "release_id": application["release_id"],
                "target_id": application["target_id"],
                "prompt_sha256": prompt_digest,
                "adapter_sha256": adapter_digest,
                "consumers": expected_consumers,
            }
            if application_observed == exact_application_state:
                break
            time.sleep(float(positive(runtime["poll_interval_seconds"], "poll interval")))
        else:
            raise TimeoutError("application-owned prompt/adapter state did not converge on every declared consumer")

    receipt = {"schema_version": "aidefend.r001001-hosted-readback/v1", "result": "PASS", "release_id": manifest["release_id"], "incident_id": manifest["incident_id"], "manifest_sha256": hashlib.sha256(raw_manifest).hexdigest(), "deployment_resource_id": expected_id, "provider_readback_sha256": hashlib.sha256(canonical(observed)).hexdigest(), "application_target_id": application["target_id"], "application_effective_state_sha256": hashlib.sha256(canonical(application_observed)).hexdigest(), "consumer_ids": consumers, "prompt_bundle_sha256": prompt_digest, "adapter_bundle_sha256": adapter_digest}
    with args.output.open("xb") as target:
        target.write(canonical(receipt) + b"\n")
        target.flush()
        os.fsync(target.fileno())
    print(json.dumps(receipt, sort_keys=True))


if __name__ == "__main__":
    main()

Run this under a workload identity limited to the named deployment. The token exists only inside the Python process and HTTP header; it is never placed in a shell variable, command line, subprocess argument, output, or log. The Azure PUT/GET proves only provider-owned model-deployment fields. The separately authenticated application adapter applies the prompt and adapter locators to the exact non-serving recovery release, then reads every declared consumer and requires each one to report the same loaded byte digests before the combined receipt can be PASS. A separate release verifier re-runs manifest/bundle and artifact-digest checks, performs both authoritative readbacks, compares the complete consumer population, and signs the result. Test a sibling target or release, a missing or stale replica, prompt/adapter fetch failure, partial provider or application apply, retry after timeout, redirect, wrong origin/API version, duplicate keys, non-finite values, oversized bodies, stale or changing manifest bytes, and every readback mismatch. Keep the recovery release non-serving on failure; Azure control-plane success alone never authorizes cutover.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-R-001.001: Versioned Model Rollback & Restoration | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service