AID-R-002: Data Integrity Recovery for AI Systems

Version 1.20261015 · Actionable technique

Restore the integrity of source datasets used by or generated by AI systems that were corrupted, tampered with, or maliciously altered during a security incident. This includes training data, validation data, feature-store sources, source corpora that feed RAG systems, configuration data, or logs of AI outputs. Recovery typically involves reverting to known-good backups, using data validation tools to identify and correct inconsistencies, or, in some cases, reconstructing data if backups are insufficient or also compromised.

Pillars: data

Lifecycle: improvement

Scope boundary

Owns restoration or deterministic repair of trusted source records and datasets after incident corruption, including source corpora that feed retrieval pipelines. It identifies downstream impact but does not roll back, quarantine, rebuild, or cut over a vector index namespace.

AID-R-005: AID-R-002 restores the source records that feed retrieval; AID-R-005 restores the derived vector-index serving artifact. Source-data recovery does not remove tainted vectors, publish anti-reingestion tombstones, rebuild a clean index, or verify the live vector alias cutover.

Implementation guidance

AID-R-002-G001

Identify all affected data stores, downstream assets, and derived artifacts.
Purpose and boundary

Use this reusable lineage-reconciliation module to verify that the recovery authority has identified the complete, incident-bound population of affected data stores, downstream assets, and derived artifacts. It discovers and records scope; it does not restore data or mutate vector indexes.

Bind the exact signed scope

Start from an authority-signed incident seed manifest that binds the system or tenant, incident ID, trusted cutoff, immutable identifiers and digests for every confirmed compromised root, allowed lineage-source snapshots, connector releases, asset-type policy, and expected connector population. A ticket list, caller-supplied root, or catalog search result alone is not authoritative scope.

# File: recovery/contracts/blast-radius-contract.yaml
schema_version: aidefend.r002.blast-radius-contract.v1
scope_required_fields:
  - system_or_tenant
  - incident_id
  - trusted_cutoff_at
  - compromised_root_manifest_digest
  - lineage_source_manifest_digest
  - connector_release_digests
  - asset_type_policy_digest
  - expected_connector_ids
edge_required_fields:
  - source_asset_id
  - source_version_or_digest
  - target_asset_id
  - target_version_or_digest
  - relation_type
  - run_or_event_id
  - observed_at
result_required_fields:
  - expected_root_count
  - observed_root_count
  - expected_connector_count
  - observed_connector_count
  - affected_asset_ids
  - unresolved_asset_ids
  - affected_vector_index_ids_for_r005
  - canonical_edge_set_digest
  - result
Enumerate, reconcile, and retain evidence

Use read-only adapters to export immutable snapshots from the governed catalog, feature store, pipeline/run registry, object or database inventory, and RAG build registry named by the manifest. Normalize stable asset IDs and traverse every directed lineage edge from every seed. Preserve missing partitions, unsupported relation types, duplicate edges, cycles, unresolved aliases, and connector errors instead of dropping them. The owner signs the canonical node and edge sets and stores the raw exports, manifests, logs, and detached signature in retention-locked incident evidence. Vector-index IDs are a signed handoff to AID-R-005, never proof that R-005 completed.

Independent replay

A verifier identity unable to change catalogs, connectors, or owner evidence independently fetches every signed source snapshot, verifies its digest and freshness, reruns normalization and traversal with a pinned verifier release, and compares exact root, connector, node, edge, unresolved, and handoff populations. Dependencies are the signed incident scope, authoritative source inventory, immutable exports, trusted time, connector trust roots, and retention store; verify every upstream receipt and scope binding locally.

Action: release the signed blast-radius receipt only after exact population reconciliation and independent replay; never infer completeness from the assets one catalog happened to return.

AID-R-002-G002

Restore each affected data store from the most recent trusted, pre-incident backup or snapshot.
Purpose and boundary

Use this store-adapter reference architecture to verify that every affected data asset selected by the recovery authority was restored into an isolated target from the exact trusted, pre-incident immutable recovery point. It does not validate semantic fitness for production; AID-R-002-G004 owns that decision.

Authorize one exact restoration plan

Consume the signed AID-R-002-G001 scope receipt only as population input. The recovery authority must sign a plan that binds each asset ID and type, source backup locator and immutable version, source digest or complete member manifest, retention/readability proof, trusted cutoff, encryption-key version, store-adapter release, isolated target identity, expected object/table/partition population, and validation handoff. A "latest" pointer, mutable alias, or operator-selected console history is prohibited.

# File: recovery/contracts/data-restore-plan.yaml
schema_version: aidefend.r002.data-restore-plan.v1
plan_required_fields:
  - system_or_tenant
  - incident_id
  - recovery_release_id
  - trusted_cutoff_at
  - blast_radius_receipt_digest
  - recovery_policy_digest
asset_required_fields:
  - asset_id
  - asset_kind
  - source_backup_immutable_locator
  - source_version_or_snapshot_id
  - source_manifest_digest
  - source_created_at
  - encryption_key_version
  - adapter_release_digest
  - isolated_target_id
  - expected_member_population_digest
receipt_required_fields:
  - expected_asset_count
  - restored_asset_count
  - expected_member_count
  - observed_member_count
  - per_asset_readback_digests
  - failed_asset_ids
  - result
Restore, read back, and retain evidence

A recovery operator verifies the authority signature, backup retention/readability, source time against the cutoff, and every member digest before invoking the pinned store adapter. Restore only into a newly created isolated namespace or account; never copy over the serving dataset. The owner receipt records adapter logs, source and target identifiers, all expected/observed member keys, bytes and digests, key versions, start/end times, and failures. Store the canonical plan, raw logs, receipt, and detached signatures in retention-locked evidence.

Independent verification

A read-only verifier outside the recovery-operator trust domain enumerates the isolated target from its control plane, re-fetches a policy-defined complete member population, recomputes digests and counts, and confirms no source newer than the trusted cutoff was used. Dependencies are a verified blast-radius receipt, signed recovery plan and policy, immutable readable backups, adapter trust root, isolated target, encryption access, trusted time, and evidence store. Verify every dependency locally before accepting the restored asset.

Action: keep the restored target isolated and hand its immutable receipt to validation; a successful copy command or healthy database process is never restore proof.

AID-R-002-G003

If no clean backup exists, deterministically repair or reconstruct the immutable damaged source into a new isolated candidate using explicit rules and validation logic; never mutate the damaged or serving store in place.
Purpose and boundary

Use this deterministic repair reference architecture for the no-clean-backup path to verify that every input record or object was classified exactly once by an approved, incident-specific repair ruleset and that a new immutable candidate plus a complete decision ledger were produced. It never edits the serving store in place.

Sign the repair population and executable rules

The recovery authority signs the immutable damaged-source locator and manifest, exact expected record/object keys, incident evidence digests supporting each repair rule, rules package and container digests, schema and canonicalization versions, permitted outcomes, reviewer identities, and output namespace. Thresholds, allowlists, transformations, and exclusions come only from the versioned organizational repair policy.

# File: recovery/contracts/data-repair-plan.yaml
schema_version: aidefend.r002.data-repair-plan.v1
required_inputs:
  - system_or_tenant
  - incident_id
  - damaged_source_manifest_digest
  - expected_record_key_manifest_digest
  - repair_policy_digest
  - repair_rules_package_digest
  - execution_image_digest
  - schema_digest
  - output_namespace
decision_values:
  - KEEP
  - DROP
  - TRANSFORM
decision_required_fields:
  - stable_record_key
  - input_digest
  - decision
  - rule_id
  - incident_evidence_digest
  - output_digest
result_required_fields:
  - expected_record_count
  - observed_record_count
  - kept_count
  - dropped_count
  - transformed_count
  - unclassified_count
  - decision_ledger_digest
  - candidate_manifest_digest
  - result
Execute deterministically and preserve evidence

Run the digest-pinned repair package in a network-denied, read-only-source job with fixed locale, canonical serialization, and policy-supplied seeds wherever an algorithm requires randomness. Reject duplicate keys, missing inputs, undeclared rules, and any record that cannot be classified. Write the candidate to a new immutable namespace and emit a row/object decision ledger; never silently discard parse failures or overwrite the damaged source. The owner signs and retains the plan, inputs, rules, logs, ledger, output manifest, counts, and errors.

Independent replay

An independent verifier re-fetches the same immutable source and rules, verifies approvals and digests, reruns the complete population in a separate environment, and compares every decision and output digest. It also runs signed golden positive/negative cases and confirms the serving store was not mutated. Dependencies are the signed no-clean-backup decision, exact source population, incident evidence, approved rules and reviewers, isolated compute, trusted time, and evidence retention; dependency success is only eligibility.

Action: treat the repaired candidate as provisional input to AID-R-002-G004; repair completion alone never authorizes production reuse.

AID-R-002-G004

Re-validate integrity, schema, and statistical consistency of restored/repaired data before reuse.
Purpose and boundary

Use this reusable restored-data validation module to verify that the exact restored or repaired candidate population satisfies the signed integrity, schema, provenance, and statistical-consistency policy for its bound asset version. It is executed by an evaluator independent of the restorer or repair job.

Bind policy, candidate, and complete population

The recovery authority signs a validation manifest containing the incident and recovery release, candidate immutable locator and manifest digest, expected record/object/partition keys, schema and provenance-policy digests, baseline asset/version and profile digest, metric algorithms, cohorts, exclusions, organization-approved thresholds, evaluator image digest, and freshness requirements. No threshold, range, drift tolerance, sample count, or exception is embedded as a framework default.

# File: recovery/contracts/restored-data-validation.yaml
schema_version: aidefend.r002.restored-data-validation.v1
required_bindings:
  - system_or_tenant
  - incident_id
  - recovery_release_id
  - candidate_manifest_digest
  - expected_population_digest
  - schema_digest
  - provenance_policy_digest
  - baseline_asset_digest
  - baseline_profile_digest
  - metric_policy_digest
  - evaluator_image_digest
result_required_fields:
  - expected_record_count
  - observed_record_count
  - measured_record_count
  - excluded_record_count
  - errored_record_count
  - integrity_result_digest
  - schema_result_digest
  - provenance_result_digest
  - statistical_result_digest
  - failed_rule_ids
  - result
Measure and retain evidence

Fetch the candidate and baseline by immutable digest, reconcile every expected key before metric computation, verify provenance and source bindings, validate the signed schema, and calculate each policy metric over its exact cohort. Preserve raw per-record/rule results, exclusions, parse failures, and metric inputs. The evaluator signs the canonical verdict and stores the manifest, raw outputs, receipt, signature, and immutable candidate readback in a retention-locked store.

Independent replay

A second read-only assurance identity verifies every input signature and digest, independently enumerates the candidate, recomputes integrity/schema/provenance checks and metrics using a pinned implementation, and compares row results and verdict. Dependencies are the signed recovery candidate receipt, validation and metric policies, trusted baseline, complete population manifest, evaluator trust roots, trusted time, and evidence store; neither restoration nor repair PASS becomes validation PASS.

Action: issue the validation receipt only for the exact immutable candidate; any later mutation, new partition, or changed policy requires a new validation run.

AID-R-002-G005

Keep recovered data blocked from production reuse until the incident's exact ingestion or update path has an authenticated remediation record and an independent read-only path verifies the effective gate configuration and behavior.
Purpose and boundary

This is the final data-reuse cutover gate for the recovered artifact. AID-R-002 does not implement or score the underlying ingestion, update, provenance, or authorization fix. The relevant Harden control or AID-R-004 owns that remediation. This guidance keeps the exact recovered dataset blocked until its incident path has a valid owner receipt and an independent verifier proves that the effective production gate now rejects the original corruption case while accepting an authorized clean control.

Bind the cutover to the recovered bytes and exact incident path

The recovery authority signs a cutover manifest that binds the incident and tenant, recovered artifact digest, the exact signed source-snapshot and AID-R-002-G004 validation-receipt payload/bundle digests, recovery release, source and candidate populations, exact ingestion or update path identifier, required owner-control ID, owner- and verifier-receipt contracts, expected effective-gate digest and generation, and expiry. A generic ticket, deployment completion event, or unrelated control PASS cannot authorize reuse.

{
  "schema_version": "aidefend.r002.data-reuse-cutover.v1",
  "control": "AID-R-002",
  "system_or_tenant": "example-system",
  "incident_id": "INC-EXAMPLE",
  "recovery_release_id": "recovery-example",
  "recovered_artifact_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "source_snapshot_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
  "source_population_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
  "candidate_manifest_sha256": "2222222222222222222222222222222222222222222222222222222222222222",
  "integrity_validation_receipt_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
  "expected_population_sha256": "3333333333333333333333333333333333333333333333333333333333333333",
  "incident_path_id": "ingestion/customer-feedback/v3",
  "required_owner_control": "AID-H-003.003",
  "required_owner_receipt_contract": "aidefend.h003003.owner-receipt.v1",
  "required_verifier_receipt_contract": "aidefend.h003003.verification.v1",
  "expected_effective_gate_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
  "expected_effective_gate_generation": "gate-generation-42",
  "expires_at": "2026-07-21T00:00:00Z"
}
Verify payload/bundle pairs, scope, and effective behavior before admission

The deployment wrapper pins three distinct trust roots under /etc/aidefend/r002/trust. The evaluator rejects a symlink, non-regular key, non-root ownership, or group/world-writable mode, captures each key once through the protected directory, and verifies only against the private key snapshot; evidence-path callers cannot choose a trust root.

set -euo pipefail
set -o noclobber
umask 077
python recovery/build_r002_cutover_input.py \
  --manifest recovery/r002-data-reuse-cutover.json \
  --manifest-bundle recovery/r002-data-reuse-cutover.json.sigstore.json \
  --source-snapshot evidence/source-snapshot.json \
  --source-snapshot-bundle evidence/source-snapshot.json.sigstore.json \
  --validation-receipt evidence/r002-validation-receipt.json \
  --validation-bundle evidence/r002-validation-receipt.json.sigstore.json \
  --owner-receipt evidence/owner-receipt.json \
  --owner-bundle evidence/owner-receipt.json.sigstore.json \
  --verifier-receipt evidence/verification-receipt.json \
  --verifier-bundle evidence/verification-receipt.json.sigstore.json \
  --recovered-artifact recovery/recovered-dataset.bin \
  --output evidence/cutover-input.json \
  | conftest test - --parser json \
      --namespace aidefend.r002.data_reuse_cutover \
      --policy policy/r002-data-reuse-cutover.rego
conftest test evidence/cutover-input.json --parser json \
  --namespace aidefend.r002.data_reuse_cutover \
  --policy policy/r002-data-reuse-cutover.rego
fixture_root="$(mktemp -d)"
trap 'rm -rf -- "$fixture_root"' EXIT
jq '.verifier_receipt.negative_incident_replay = "ALLOWED"' \
  evidence/cutover-input.json > "$fixture_root/rejected.json"
if conftest test "$fixture_root/rejected.json" --parser json \
    --namespace aidefend.r002.data_reuse_cutover \
    --policy policy/r002-data-reuse-cutover.rego; then
  echo "negative cutover fixture unexpectedly passed" >&2
  exit 1
fi

# File: recovery/build_r002_cutover_input.py
from __future__ import annotations

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

SHA256 = re.compile(r"^[0-9a-f]{64}$")
MAXIMUM_SIGNED_DOCUMENT_BYTES = int(os.environ["R002_MAXIMUM_SIGNED_DOCUMENT_BYTES"])
MAXIMUM_RECOVERED_ARTIFACT_BYTES = int(os.environ["R002_MAXIMUM_RECOVERED_ARTIFACT_BYTES"])
SIGNATURE_TIMEOUT_SECONDS = float(os.environ["R002_SIGNATURE_TIMEOUT_SECONDS"])
MAXIMUM_EVIDENCE_AGE_SECONDS = float(os.environ["R002_MAXIMUM_EVIDENCE_AGE_SECONDS"])
MAXIMUM_FUTURE_SKEW_SECONDS = float(os.environ["R002_MAXIMUM_FUTURE_SKEW_SECONDS"])
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
TRUST_ROOT = Path("/etc/aidefend/r002/trust")
MANIFEST_KEY = TRUST_ROOT / "recovery-authority.pub"
OWNER_KEY = TRUST_ROOT / "ingestion-owner.pub"
VERIFIER_KEY = TRUST_ROOT / "independent-verifier.pub"
if (
    MAXIMUM_SIGNED_DOCUMENT_BYTES <= 0
    or MAXIMUM_RECOVERED_ARTIFACT_BYTES <= 0
    or any(
        not math.isfinite(value) or value <= 0
        for value in (
            SIGNATURE_TIMEOUT_SECONDS,
            MAXIMUM_EVIDENCE_AGE_SECONDS,
        )
    )
    or not math.isfinite(MAXIMUM_FUTURE_SKEW_SECONDS)
    or MAXIMUM_FUTURE_SKEW_SECONDS < 0
    or not RUNTIME_PROFILE_VERSION
    or SHA256.fullmatch(RUNTIME_PROFILE_SHA256) is None
):
    raise RuntimeError("versioned R002 cutover runtime profile is invalid")

MANIFEST_FIELDS = {
    "schema_version", "control", "system_or_tenant", "incident_id",
    "recovery_release_id", "recovered_artifact_sha256", "source_snapshot_sha256",
    "source_population_sha256", "candidate_manifest_sha256",
    "integrity_validation_receipt_sha256", "expected_population_sha256", "incident_path_id",
    "required_owner_control", "required_owner_receipt_contract",
    "required_verifier_receipt_contract", "expected_effective_gate_sha256",
    "expected_effective_gate_generation", "expires_at",
}
SOURCE_REQUIRED_FIELDS = {
    "schema_version", "system_or_tenant", "incident_id", "recovery_release_id",
    "source_population_sha256", "captured_at",
}
VALIDATION_REQUIRED_FIELDS = {
    "schema_version", "system_or_tenant", "incident_id", "recovery_release_id",
    "candidate_manifest_sha256", "expected_population_sha256",
    "recovered_artifact_sha256", "completed_at", "valid_until", "result",
}
OWNER_FIELDS = {
    "schema_version", "control", "system_or_tenant", "incident_id",
    "incident_path_id", "recovered_artifact_sha256",
    "effective_gate_deployment_sha256", "effective_gate_generation", "completed_at", "result",
}
VERIFIER_FIELDS = {
    "schema_version", "system_or_tenant", "incident_id", "incident_path_id",
    "recovered_artifact_sha256", "source_snapshot_sha256",
    "integrity_validation_receipt_sha256", "owner_receipt_sha256",
    "effective_gate_sha256", "effective_gate_generation", "negative_incident_replay",
    "positive_clean_replay", "observed_at", "valid_until", "result",
}


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_object(raw: bytes, label: str) -> 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(f"{label} must contain one JSON object")
    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 versioned runtime 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 capture_trust_key(path: Path) -> bytes:
    if path.parent != TRUST_ROOT:
        raise PermissionError("trust key is outside the fixed trust root")
    root_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    root_descriptor = os.open(TRUST_ROOT, root_flags)
    try:
        root_status = os.fstat(root_descriptor)
        if (
            not stat.S_ISDIR(root_status.st_mode)
            or root_status.st_uid != 0
            or root_status.st_mode & 0o022
        ):
            raise PermissionError("trust root is not root-owned and protected")
        descriptor = os.open(
            path.name,
            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0),
            dir_fd=root_descriptor,
        )
        try:
            before = os.fstat(descriptor)
            if (
                not stat.S_ISREG(before.st_mode)
                or before.st_uid != 0
                or before.st_mode & 0o022
                or before.st_size < 1
                or before.st_size > MAXIMUM_SIGNED_DOCUMENT_BYTES
            ):
                raise PermissionError("trust key is not a bounded root-owned protected file")
            chunks = []
            while True:
                block = os.read(descriptor, 64 * 1024)
                if not block:
                    break
                chunks.append(block)
            after = os.fstat(descriptor)
        finally:
            os.close(descriptor)
    finally:
        os.close(root_descriptor)
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size,
        item.st_mtime_ns, item.st_ctime_ns,
    )
    raw = b"".join(chunks)
    if identity(before) != identity(after) or len(raw) != before.st_size:
        raise RuntimeError("trust key changed during capture")
    return raw


def verified_object(
    path: Path,
    bundle: Path,
    key_bytes: bytes,
    snapshot_name: str,
) -> tuple[dict[str, Any], bytes]:
    raw = read_limited(path, MAXIMUM_SIGNED_DOCUMENT_BYTES)
    bundle_raw = read_limited(bundle, MAXIMUM_SIGNED_DOCUMENT_BYTES)
    if not raw or not bundle_raw:
        raise ValueError("signed payload and Sigstore bundle must both be non-empty")
    with tempfile.TemporaryDirectory(prefix="r002-cutover-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        payload_snapshot = snapshot_root / snapshot_name
        bundle_snapshot = snapshot_root / (snapshot_name + ".sigstore.json")
        key_snapshot = snapshot_root / (snapshot_name + ".pub")
        write_private_snapshot(payload_snapshot, raw)
        write_private_snapshot(bundle_snapshot, bundle_raw)
        write_private_snapshot(key_snapshot, key_bytes)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(key_snapshot), "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=SIGNATURE_TIMEOUT_SECONDS,
        )
        if read_limited(payload_snapshot, MAXIMUM_SIGNED_DOCUMENT_BYTES) != raw:
            raise RuntimeError("verified R002 payload snapshot changed")
        if read_limited(bundle_snapshot, MAXIMUM_SIGNED_DOCUMENT_BYTES) != bundle_raw:
            raise RuntimeError("verified R002 Sigstore bundle snapshot changed")
        if read_limited(key_snapshot, MAXIMUM_SIGNED_DOCUMENT_BYTES) != key_bytes:
            raise RuntimeError("verified R002 trust-key snapshot changed")
    return strict_object(raw, str(path)), raw


def digest(value: object, field: str) -> str:
    if (
        not isinstance(value, str)
        or SHA256.fullmatch(value) is None
        or set(value) == {"0"}
    ):
        raise ValueError(f"{field} must be a nonzero lowercase SHA-256 digest")
    return value


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


def timestamp(value: object, field: str) -> datetime:
    try:
        parsed = datetime.fromisoformat(text(value, field).replace("Z", "+00:00"))
    except ValueError as error:
        raise ValueError(f"{field} must be RFC 3339") from error
    if parsed.tzinfo is None:
        raise ValueError(f"{field} must include a timezone")
    return parsed.astimezone(timezone.utc)


def stable_artifact_digest(path: Path) -> str:
    observed = 0
    hasher = hashlib.sha256()
    with path.open("rb") as source:
        before = os.fstat(source.fileno())
        for chunk in iter(lambda: source.read(1024 * 1024), b""):
            observed += len(chunk)
            if observed > MAXIMUM_RECOVERED_ARTIFACT_BYTES:
                raise ValueError("recovered artifact exceeds the versioned bound")
            hasher.update(chunk)
        after = os.fstat(source.fileno())
    if observed != before.st_size or (
        before.st_size,
        before.st_mtime_ns,
    ) != (
        after.st_size,
        after.st_mtime_ns,
    ):
        raise RuntimeError("recovered artifact changed during hashing")
    return hasher.hexdigest()


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


def read_existing(directory_descriptor: int, name: str, maximum: int) -> bytes | None:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(name, flags, dir_fd=directory_descriptor)
    except FileNotFoundError:
        return None
    with os.fdopen(descriptor, "rb") as source:
        raw = source.read(maximum + 1)
        status = os.fstat(source.fileno())
    if not stat.S_ISREG(status.st_mode) or len(raw) > maximum:
        raise FileExistsError("cutover output is not one bounded regular file")
    return raw


def publish_once(path: Path, payload: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    parent = path.parent.resolve(strict=True)
    if path.parent.absolute() != parent or path.name in {"", ".", ".."}:
        raise PermissionError("cutover output parent must be one direct, non-symlink directory")
    directory_flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
    directory_descriptor = os.open(parent, directory_flags)
    staging_name = "." + path.name + "." + secrets.token_hex(12) + ".tmp"
    try:
        prior = read_existing(directory_descriptor, path.name, len(payload))
        if prior is not None:
            if prior != payload:
                raise FileExistsError("different cutover evidence already exists")
            return
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
        staging_descriptor = os.open(staging_name, flags, 0o600, dir_fd=directory_descriptor)
        try:
            with os.fdopen(staging_descriptor, "wb", closefd=False) as target:
                target.write(payload)
                target.flush()
                os.fsync(target.fileno())
        finally:
            os.close(staging_descriptor)
        try:
            os.link(staging_name, path.name, src_dir_fd=directory_descriptor, dst_dir_fd=directory_descriptor, follow_symlinks=False)
        except FileExistsError:
            prior = read_existing(directory_descriptor, path.name, len(payload))
            if prior != payload:
                raise
        os.fsync(directory_descriptor)
    finally:
        try:
            os.unlink(staging_name, dir_fd=directory_descriptor)
        except FileNotFoundError:
            pass
        os.close(directory_descriptor)


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("--source-snapshot", type=Path, required=True)
    parser.add_argument("--source-snapshot-bundle", type=Path, required=True)
    parser.add_argument("--validation-receipt", type=Path, required=True)
    parser.add_argument("--validation-bundle", type=Path, required=True)
    parser.add_argument("--owner-receipt", type=Path, required=True)
    parser.add_argument("--owner-bundle", type=Path, required=True)
    parser.add_argument("--verifier-receipt", type=Path, required=True)
    parser.add_argument("--verifier-bundle", type=Path, required=True)
    parser.add_argument("--recovered-artifact", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    for payload, bundle, label in ((args.manifest, args.manifest_bundle, "cutover manifest"), (args.source_snapshot, args.source_snapshot_bundle, "source snapshot"), (args.validation_receipt, args.validation_bundle, "validation receipt"), (args.owner_receipt, args.owner_bundle, "owner receipt"), (args.verifier_receipt, args.verifier_bundle, "verifier receipt")):
        if bundle != Path(str(payload) + ".sigstore.json"):
            raise ValueError(f"{label} bundle must be the sibling <payload>.sigstore.json")

    manifest_key = capture_trust_key(MANIFEST_KEY)
    owner_key = capture_trust_key(OWNER_KEY)
    verifier_key = capture_trust_key(VERIFIER_KEY)
    if len({hashlib.sha256(key).digest() for key in (manifest_key, owner_key, verifier_key)}) != 3:
        raise PermissionError("manifest, owner, and verifier trust keys must be distinct")

    manifest, manifest_bytes = verified_object(
        args.manifest, args.manifest_bundle, manifest_key, "cutover-manifest.json"
    )
    source_snapshot, source_snapshot_bytes = verified_object(
        args.source_snapshot, args.source_snapshot_bundle, manifest_key, "source-snapshot.json"
    )
    validation, validation_bytes = verified_object(
        args.validation_receipt, args.validation_bundle, verifier_key, "validation-receipt.json"
    )
    owner, owner_bytes = verified_object(
        args.owner_receipt, args.owner_bundle, owner_key, "owner-receipt.json"
    )
    verifier, verifier_bytes = verified_object(
        args.verifier_receipt,
        args.verifier_bundle,
        verifier_key,
        "verification-receipt.json",
    )
    if (
        set(manifest) != MANIFEST_FIELDS
        or manifest.get("schema_version") != "aidefend.r002.data-reuse-cutover.v1"
        or manifest.get("control") != "AID-R-002"
    ):
        raise ValueError("R002 cutover manifest schema differs")
    if not SOURCE_REQUIRED_FIELDS.issubset(source_snapshot) or not VALIDATION_REQUIRED_FIELDS.issubset(validation):
        raise ValueError("source snapshot or validation receipt schema differs")
    if set(owner) != OWNER_FIELDS or set(verifier) != VERIFIER_FIELDS:
        raise ValueError("owner or verifier receipt schema differs")
    for field in (
        "system_or_tenant",
        "incident_id",
        "recovery_release_id",
        "incident_path_id",
        "required_owner_control",
        "required_owner_receipt_contract",
        "required_verifier_receipt_contract",
        "expected_effective_gate_generation",
    ):
        text(manifest.get(field), f"manifest.{field}")
    for field in (
        "recovered_artifact_sha256",
        "source_snapshot_sha256",
        "source_population_sha256",
        "candidate_manifest_sha256",
        "integrity_validation_receipt_sha256",
        "expected_population_sha256",
        "expected_effective_gate_sha256",
    ):
        digest(manifest.get(field), f"manifest.{field}")

    if hashlib.sha256(source_snapshot_bytes).hexdigest() != manifest["source_snapshot_sha256"] or hashlib.sha256(validation_bytes).hexdigest() != manifest["integrity_validation_receipt_sha256"]:
        raise RuntimeError("source snapshot or validation receipt bytes differ from the cutover manifest")
    source_scope = (
        source_snapshot.get("system_or_tenant"),
        source_snapshot.get("incident_id"),
        source_snapshot.get("recovery_release_id"),
        source_snapshot.get("source_population_sha256"),
    )
    validation_scope = (
        validation.get("system_or_tenant"),
        validation.get("incident_id"),
        validation.get("recovery_release_id"),
        validation.get("candidate_manifest_sha256"),
        validation.get("expected_population_sha256"),
        validation.get("recovered_artifact_sha256"),
    )
    if source_snapshot.get("schema_version") != "aidefend.r002.source-snapshot.v1" or source_scope != (manifest["system_or_tenant"], manifest["incident_id"], manifest["recovery_release_id"], manifest["source_population_sha256"]):
        raise RuntimeError("source snapshot subject or population differs")
    if validation.get("schema_version") != "aidefend.r002.restored-data-validation-receipt.v1" or validation.get("result") != "PASS" or validation_scope != (manifest["system_or_tenant"], manifest["incident_id"], manifest["recovery_release_id"], manifest["candidate_manifest_sha256"], manifest["expected_population_sha256"], manifest["recovered_artifact_sha256"]):
        raise RuntimeError("G004 validation subject, population, or result differs")
    expected_scope = (
        manifest["system_or_tenant"],
        manifest["incident_id"],
        manifest["incident_path_id"],
        manifest["recovered_artifact_sha256"],
    )
    owner_scope = (
        owner.get("system_or_tenant"),
        owner.get("incident_id"),
        owner.get("incident_path_id"),
        owner.get("recovered_artifact_sha256"),
    )
    verifier_scope = (
        verifier.get("system_or_tenant"),
        verifier.get("incident_id"),
        verifier.get("incident_path_id"),
        verifier.get("recovered_artifact_sha256"),
    )
    if owner_scope != expected_scope or verifier_scope != expected_scope:
        raise RuntimeError("cutover receipt scope differs")
    if (
        owner.get("schema_version") != manifest["required_owner_receipt_contract"]
        or owner.get("control") != manifest["required_owner_control"]
        or owner.get("result") != "PASS"
        or digest(
            owner.get("effective_gate_deployment_sha256"),
            "owner.effective_gate_deployment_sha256",
        )
        != manifest["expected_effective_gate_sha256"]
        or owner.get("effective_gate_generation") != manifest["expected_effective_gate_generation"]
    ):
        raise RuntimeError("owner remediation receipt differs")
    if (
        verifier.get("schema_version")
        != manifest["required_verifier_receipt_contract"]
        or verifier.get("source_snapshot_sha256") != manifest["source_snapshot_sha256"]
        or verifier.get("integrity_validation_receipt_sha256") != manifest["integrity_validation_receipt_sha256"]
        or verifier.get("owner_receipt_sha256")
        != hashlib.sha256(owner_bytes).hexdigest()
        or verifier.get("effective_gate_sha256")
        != manifest["expected_effective_gate_sha256"]
        or verifier.get("effective_gate_generation") != manifest["expected_effective_gate_generation"]
        or verifier.get("negative_incident_replay") != "DENIED"
        or verifier.get("positive_clean_replay") != "ALLOWED"
        or verifier.get("result") != "PASS"
    ):
        raise RuntimeError("independent effective-gate verification differs")
    if stable_artifact_digest(args.recovered_artifact) != manifest["recovered_artifact_sha256"]:
        raise RuntimeError("recovered artifact bytes differ from the cutover manifest")

    now = datetime.now(timezone.utc)
    source_captured_at = timestamp(source_snapshot.get("captured_at"), "source_snapshot.captured_at")
    validation_completed_at = timestamp(validation.get("completed_at"), "validation.completed_at")
    validation_valid_until = timestamp(validation.get("valid_until"), "validation.valid_until")
    completed_at = timestamp(owner.get("completed_at"), "owner.completed_at")
    observed_at = timestamp(verifier.get("observed_at"), "verifier.observed_at")
    valid_until = timestamp(verifier.get("valid_until"), "verifier.valid_until")
    expires_at = timestamp(manifest.get("expires_at"), "manifest.expires_at")
    for label, observed in (("source_snapshot.captured_at", source_captured_at), ("validation.completed_at", validation_completed_at), ("owner.completed_at", completed_at), ("verifier.observed_at", observed_at)):
        age = (now - observed).total_seconds()
        if age > MAXIMUM_EVIDENCE_AGE_SECONDS or age < -MAXIMUM_FUTURE_SKEW_SECONDS:
            raise RuntimeError(f"{label} is stale or future-dated")
    if observed_at < completed_at or observed_at >= valid_until:
        raise RuntimeError("independent verifier observation is inverted or outside its validity interval")
    if validation_completed_at > now or now >= validation_valid_until:
        raise RuntimeError("G004 validation receipt is future-dated or no longer current")
    if now >= valid_until or now >= expires_at:
        raise RuntimeError("verifier receipt or cutover manifest expired")

    manifest_for_policy = dict(manifest)
    manifest_for_policy["expires_at_unix"] = expires_at.timestamp()
    cutover_input = {
        "manifest": manifest_for_policy,
        "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
        "source_snapshot": source_snapshot,
        "source_snapshot_sha256": hashlib.sha256(source_snapshot_bytes).hexdigest(),
        "validation_receipt": validation,
        "validation_receipt_sha256": hashlib.sha256(validation_bytes).hexdigest(),
        "owner_receipt": owner,
        "owner_receipt_sha256": hashlib.sha256(owner_bytes).hexdigest(),
        "verifier_receipt": verifier,
        "verifier_receipt_sha256": hashlib.sha256(verifier_bytes).hexdigest(),
        "recovered_artifact_sha256": manifest["recovered_artifact_sha256"],
        "now_unix": now.timestamp(),
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
    }
    payload = canonical(cutover_input)
    publish_once(args.output, payload)
    sys.stdout.buffer.write(payload)


if __name__ == "__main__":
    main()

package aidefend.r002.data_reuse_cutover

default allow := false

allow if {
  input.manifest.control == "AID-R-002"
  input.manifest.recovered_artifact_sha256 == input.recovered_artifact_sha256
  input.owner_receipt.schema_version == input.manifest.required_owner_receipt_contract
  input.owner_receipt.control == input.manifest.required_owner_control
  input.owner_receipt.system_or_tenant == input.manifest.system_or_tenant
  input.owner_receipt.incident_id == input.manifest.incident_id
  input.owner_receipt.incident_path_id == input.manifest.incident_path_id
  input.owner_receipt.recovered_artifact_sha256 == input.recovered_artifact_sha256
  input.owner_receipt.result == "PASS"
  input.owner_receipt.effective_gate_generation == input.manifest.expected_effective_gate_generation
  input.source_snapshot_sha256 == input.manifest.source_snapshot_sha256
  input.validation_receipt_sha256 == input.manifest.integrity_validation_receipt_sha256
  input.validation_receipt.result == "PASS"
  input.validation_receipt.candidate_manifest_sha256 == input.manifest.candidate_manifest_sha256
  input.validation_receipt.expected_population_sha256 == input.manifest.expected_population_sha256
  input.validation_receipt.recovered_artifact_sha256 == input.manifest.recovered_artifact_sha256
  input.verifier_receipt.schema_version == input.manifest.required_verifier_receipt_contract
  input.verifier_receipt.system_or_tenant == input.manifest.system_or_tenant
  input.verifier_receipt.incident_id == input.manifest.incident_id
  input.verifier_receipt.incident_path_id == input.manifest.incident_path_id
  input.verifier_receipt.recovered_artifact_sha256 == input.recovered_artifact_sha256
  input.verifier_receipt.source_snapshot_sha256 == input.manifest.source_snapshot_sha256
  input.verifier_receipt.integrity_validation_receipt_sha256 == input.manifest.integrity_validation_receipt_sha256
  input.verifier_receipt.owner_receipt_sha256 == input.owner_receipt_sha256
  input.verifier_receipt.effective_gate_sha256 == input.manifest.expected_effective_gate_sha256
  input.verifier_receipt.effective_gate_generation == input.manifest.expected_effective_gate_generation
  input.verifier_receipt.negative_incident_replay == "DENIED"
  input.verifier_receipt.positive_clean_replay == "ALLOWED"
  input.verifier_receipt.result == "PASS"
  input.manifest.expires_at_unix > input.now_unix
}

deny contains msg if {
  not allow
  msg := "recovered data cutover contract is not satisfied"
}

The admission controller receives only payload/bundle-verified, schema-validated objects. It must also compare every tenant, release, artifact, incident-path, policy, and freshness binding; enumerate all affected pipelines from the signed lineage population; read the effective gate through a separate identity; and replay the incident-specific corrupt case plus a clean authorized control at every declared enforcement point. Missing coverage, stale evidence, an unavailable verifier, a digest mismatch, or an unsupported path keeps the recovered artifact quarantined. This path intentionally has no implicit exception branch: a risk-acceptance workflow, if a deployment exposes one, must remain a separately enforced product decision and must never be serialized here as remediation or validation PASS.

Exercise missing, altered, stale, future-dated, sibling-subject or wrong-population source/validation payloads; validation expiry; verifier observation before owner completion or outside valid_until; changed gate generation; and hard exits before, during and after staging write, file fsync, atomic link and directory fsync. Pre-existing different output, symlink substitution, concurrent publishers and partial staging fail closed; an idempotent retry may reuse only byte-identical committed evidence. Any later artifact, gate generation, policy or configuration change requires a new invocation and receipt. Action: release the recovered dataset only after the exact cutover manifest, source snapshot, current G004 validation receipt and independent effective-gate replay pass. Retain the decision, receipts, effective-state readback, test cases and admission transaction together.

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-002: Data Integrity Recovery for AI Systems | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service