AID-R-001.002: Model Retraining for Remediation
Build sanitized recovery-training inputs, execute a fresh retraining or recovery-learning workflow, and require the exact remediated model to meet signed clean-utility and incident-attack promotion thresholds before returning to service when rollback to a trusted prior version is unsafe or unavailable.
Pillars: model
Lifecycle: improvement
Scope boundary
Owns construction, validation, and promotion of a newly remediated model through clean retraining or recovery learning when trustworthy rollback is unavailable, incomplete, or unsafe. It does not silently substitute rollback and does not prove that the original compromised artifact or derivative was evicted.
AID-R-001.001: AID-R-001.002 creates a newly remediated model; AID-R-001.001 restores an exact trusted pre-incident version. Use retraining when rollback state is unavailable or untrustworthy. Use AID-R-001.001 when an authority-approved immutable prior version and trusted cutoff exist.
AID-E-003.005: AID-R-001.002 builds and validates the replacement model; AID-E-003.005 quarantines compromised adapters and their derivatives. For adapter-derived compromise, this control consumes signed quarantine, tombstone, and derivative-routing evidence. Retraining does not establish that the original adapter or derivative can no longer be served.
AID-I-006: AID-R-001.002 consumes the signed federated-client and update exclusion population when constructing sanitized recovery-training inputs and validating the remediated model; AID-I-006 owns the underlying quarantine decision and exclusion enforcement. Retraining must not recreate or independently score client quarantine, and quarantine alone does not prove that the replacement model is safe to promote.
Implementation guidance
- AID-R-001.002-G001: Materialize an exact signed sanitized training snapshot from authenticated eviction manifests, then submit a digest-pinned fresh recovery-training run through a trusted workflow.
- AID-R-001.002-G002: Use SISA shard retraining or exact federated-history replay as provisional remediation only when their architecture-specific prerequisites are present and a separate post-retraining validation gate is satisfied.
- AID-R-001.002-G003: For LLM incidents, run architecture-specific NPO or RMU recovery recipes only with signed forget/retain datasets and explicit provisional-release gates.
- AID-R-001.002-G004: Run a separately signed, exact-population post-remediation evaluation that checks the remediated artifact against versioned clean-utility and incident-specific attack thresholds before promotion.
AID-R-001.002-G001
Materialize an exact signed sanitized training snapshot from authenticated eviction manifests, then submit a digest-pinned fresh recovery-training run through a trusted workflow.
Verify signed snapshots before materializing recovery inputs
The source inventory, eviction/removal manifest, and training submission profile are independently signed. Read each into a private immutable snapshot, verify that exact byte sequence, reject duplicate keys/non-finite JSON, and retain the raw digest. The locally installed, versioned bootstrap runtime profile bounds trust documents, command and network deadlines, retry attempts, and response size; the verified training profile binds table, file, and row populations. A removal receipt is a dependency, not proof that the sanitized population is complete.
# File: remediation/strict_io.py
from __future__ import annotations
import hashlib
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit
import requests
INPUT_LIMIT = int(os.environ["R001_INPUT_MAXIMUM_TRUST_DOCUMENT_BYTES"])
RESPONSE_LIMIT = int(os.environ["R001_INPUT_MAXIMUM_RESPONSE_BYTES"])
CONNECT_TIMEOUT = float(os.environ["R001_INPUT_CONNECT_TIMEOUT_SECONDS"])
READ_TIMEOUT = float(os.environ["R001_INPUT_READ_TIMEOUT_SECONDS"])
COMMAND_TIMEOUT = float(os.environ["R001_INPUT_COMMAND_TIMEOUT_SECONDS"])
MAXIMUM_RETRY_ATTEMPTS = int(
os.environ["R001_INPUT_MAXIMUM_RETRY_ATTEMPTS"]
)
BOOTSTRAP_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
BOOTSTRAP_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
if (
not BOOTSTRAP_PROFILE_VERSION
or len(BOOTSTRAP_PROFILE_SHA256) != 64
or set(BOOTSTRAP_PROFILE_SHA256) - set("0123456789abcdef")
or INPUT_LIMIT < 1 or RESPONSE_LIMIT < 1
or MAXIMUM_RETRY_ATTEMPTS < 1
or any(not math.isfinite(value) or value <= 0 for value in (CONNECT_TIMEOUT, READ_TIMEOUT, COMMAND_TIMEOUT))
):
raise RuntimeError("local versioned recovery-input bootstrap profile is invalid")
TIMEOUT = (CONNECT_TIMEOUT, READ_TIMEOUT)
def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
result: dict[str, Any] = {}
for name, value in pairs:
if name in result:
raise ValueError(f"duplicate JSON key: {name}")
result[name] = value
return result
def reject_constant(value: str) -> None:
raise ValueError(f"non-finite JSON number is forbidden: {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 read_file(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_json(path: Path, bundle: Path, public_key: str) -> tuple[dict[str, Any], str]:
raw = read_file(path, INPUT_LIMIT)
bundle_raw = read_file(bundle, INPUT_LIMIT)
if not raw or not bundle_raw:
raise ValueError("signed payload and Sigstore bundle must both be non-empty")
with tempfile.TemporaryDirectory(prefix="r001002-") as directory:
snapshot_root = Path(directory)
os.chmod(snapshot_root, 0o700)
payload_snapshot = snapshot_root / "signed-input.json"
bundle_snapshot = snapshot_root / "signed-input.json.sigstore.json"
write_private_snapshot(payload_snapshot, raw)
write_private_snapshot(bundle_snapshot, bundle_raw)
subprocess.run(["cosign", "verify-blob", "--key", public_key, "--bundle", str(bundle_snapshot), str(payload_snapshot)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
if read_file(payload_snapshot, INPUT_LIMIT) != raw:
raise RuntimeError("verified JSON payload snapshot changed")
if read_file(bundle_snapshot, INPUT_LIMIT) != bundle_raw:
raise RuntimeError("verified Sigstore bundle snapshot changed")
return strict_json(raw), hashlib.sha256(raw).hexdigest()
def exact_origin(value: object) -> str:
if not isinstance(value, str):
raise ValueError("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("one exact HTTPS origin is required")
return f"https://{parsed.netloc.lower()}"
def get_json(origin: str, path: str) -> dict[str, Any]:
origin = exact_origin(origin)
session = requests.Session()
session.trust_env = False
session.cert = (os.environ["TRAINING_AUDITOR_CLIENT_CERT"], os.environ["TRAINING_AUDITOR_CLIENT_KEY"])
session.verify = os.environ["TRAINING_AUDITOR_CA_FILE"]
session.headers.update({"Accept": "application/json", "Accept-Encoding": "identity"})
with session.get(origin + path, timeout=TIMEOUT, allow_redirects=False, stream=True) as response:
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" or response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
raise RuntimeError("unexpected response encoding or media type")
chunks: list[bytes] = []
observed = 0
for chunk in response.iter_content(chunk_size=65536):
observed += len(chunk)
if observed > RESPONSE_LIMIT:
raise RuntimeError("training-auditor response exceeds the versioned bound")
chunks.append(chunk)
response.raise_for_status()
return strict_json(b"".join(chunks))
def post_json(origin: str, path: str, body: dict[str, Any], idempotency_key: str) -> dict[str, Any]:
origin = exact_origin(origin)
session = requests.Session()
session.trust_env = False
session.cert = (os.environ["TRAINING_CLIENT_CERT"], os.environ["TRAINING_CLIENT_KEY"])
session.verify = os.environ["TRAINING_CA_FILE"]
session.headers.update({"Accept": "application/json", "Accept-Encoding": "identity"})
with session.post(origin + path, data=canonical(body), headers={"Content-Type": "application/json", "Idempotency-Key": idempotency_key}, timeout=TIMEOUT, allow_redirects=False, stream=True) as response:
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" or response.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json":
raise RuntimeError("unexpected response encoding or media type")
chunks: list[bytes] = []
observed = 0
for chunk in response.iter_content(chunk_size=65536):
observed += len(chunk)
if observed > RESPONSE_LIMIT:
raise RuntimeError("training response exceeds the signed decoded-body bound")
chunks.append(chunk)
response.raise_for_status()
return strict_json(b"".join(chunks))
Materialize the exact sanitized table population and submit it
# File: remediation/materialize_and_submit_recovery.py
from __future__ import annotations
import argparse
import base64
import hashlib
import io
import json
import math
import os
import re
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Any
from urllib.parse import quote
import boto3
import pandas as pd
from botocore.config import Config
from botocore.exceptions import ClientError
from remediation.strict_io import COMMAND_TIMEOUT, CONNECT_TIMEOUT, INPUT_LIMIT, MAXIMUM_RETRY_ATTEMPTS, READ_TIMEOUT, canonical, exact_origin, get_json, post_json, read_file, verified_json
SAFE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
HEX = re.compile(r"^[0-9a-f]{64}$")
SOURCE_FIELDS = {"schema", "tenant_id", "incident_id", "tables"}
TABLE_FIELDS = {"table_id", "path", "sha256", "id_column", "column_names", "row_count", "id_population_sha256"}
REMOVAL_FIELDS = {"schema", "tenant_id", "incident_id", "source_manifest_sha256", "tables"}
REMOVAL_TABLE_FIELDS = {"table_id", "remove_ids", "remove_id_population_sha256"}
PROFILE_FIELDS = {"schema", "tenant_id", "incident_id", "sanitized_publish_bucket", "sanitized_publish_prefix", "aws_region", "training_control_origin", "training_image_digest", "training_code_digest", "initialization_policy_digest", "workflow_template", "maximum_table_count", "maximum_rows_per_table", "maximum_source_file_bytes", "recovery_input_signing_key", "recovery_input_public_key"}
def digest(value: object, name: str) -> str:
if not isinstance(value, str) or HEX.fullmatch(value) is None or set(value) == {"0"}:
raise ValueError(f"{name} must be a nonzero SHA-256 digest")
return value
def positive_int(value: object, name: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{name} must be a positive integer")
return value
def id_digest(values: list[str]) -> str:
return hashlib.sha256(canonical(values)).hexdigest()
def load_table(path: Path, spec: dict[str, Any], profile: dict[str, Any]) -> pd.DataFrame:
maximum = positive_int(profile["maximum_source_file_bytes"], "maximum source file bytes")
raw = read_file(path, maximum)
if hashlib.sha256(raw).hexdigest() != digest(spec["sha256"], "source table"):
raise RuntimeError("source table digest differs")
frame = pd.read_parquet(io.BytesIO(raw))
expected_columns = spec["column_names"]
if not isinstance(expected_columns, list) or expected_columns != list(frame.columns) or len(expected_columns) != len(set(expected_columns)):
raise RuntimeError("source table column population differs")
if len(frame) != positive_int(spec["row_count"], "source row count") or len(frame) > positive_int(profile["maximum_rows_per_table"], "maximum rows per table"):
raise RuntimeError("source table row population differs")
ids = frame[spec["id_column"]].astype(str).tolist() if spec["id_column"] in frame.columns else []
if ids != sorted(set(ids)) or id_digest(ids) != digest(spec["id_population_sha256"], "source ID population"):
raise RuntimeError("source table ID population differs")
return frame
def publish_immutable(client: Any, bucket: str, key: str, raw: bytes, maximum: int) -> str:
expected = hashlib.sha256(raw).hexdigest()
checksum = base64.b64encode(hashlib.sha256(raw).digest()).decode("ascii")
try:
client.put_object(Bucket=bucket, Key=key, Body=raw, IfNoneMatch="*", ChecksumSHA256=checksum)
except ClientError as error:
if error.response.get("ResponseMetadata", {}).get("HTTPStatusCode") not in {409, 412}:
raise
response = client.get_object(Bucket=bucket, Key=key, ChecksumMode="ENABLED")
version_id = response.get("VersionId")
observed = response["Body"].read(maximum + 1)
if len(observed) > maximum or not isinstance(version_id, str) or not version_id or hashlib.sha256(observed).hexdigest() != expected:
raise RuntimeError("immutable artifact publication/readback differs")
return "s3://" + bucket + "/" + quote(key, safe="/") + "?versionId=" + quote(version_id, safe="")
def main() -> None:
parser = argparse.ArgumentParser()
for name in ("source", "removal", "training-profile"):
parser.add_argument("--" + name, type=Path, required=True)
parser.add_argument("--" + name + "-bundle", type=Path, required=True)
parser.add_argument("--" + name + "-key", required=True)
parser.add_argument("--output-directory", type=Path, required=True)
parser.add_argument("--manifest-output", type=Path, required=True)
parser.add_argument("--bundle-output", type=Path, required=True)
args = parser.parse_args()
for payload, bundle, label in ((args.source, args.source_bundle, "source"), (args.removal, args.removal_bundle, "removal"), (args.training_profile, args.training_profile_bundle, "training profile"), (args.manifest_output, args.bundle_output, "sanitized manifest output")):
if bundle != Path(str(payload) + ".sigstore.json"):
raise ValueError(f"{label} bundle must be the sibling <payload>.sigstore.json")
source, source_raw_digest = verified_json(args.source, args.source_bundle, args.source_key)
removal, removal_raw_digest = verified_json(args.removal, args.removal_bundle, args.removal_key)
profile, profile_raw_digest = verified_json(args.training_profile, args.training_profile_bundle, args.training_profile_key)
if set(source) != SOURCE_FIELDS or set(removal) != REMOVAL_FIELDS or set(profile) != PROFILE_FIELDS or source["schema"] != "aidefend.r001002.source-tables.v1" or removal["schema"] != "aidefend.e003002.removal-manifest.v1" or profile["schema"] != "aidefend.r001002.training-profile.v1":
raise ValueError("recovery-input schema differs")
scope = (source["tenant_id"], source["incident_id"])
if (removal["tenant_id"], removal["incident_id"]) != scope or (profile["tenant_id"], profile["incident_id"]) != scope or removal["source_manifest_sha256"] != source_raw_digest:
raise RuntimeError("signed recovery inputs are not bound to one scope/snapshot")
maximum_tables = positive_int(profile["maximum_table_count"], "maximum table count")
bucket = profile["sanitized_publish_bucket"]
prefix_value = profile["sanitized_publish_prefix"]
region = profile["aws_region"]
if not isinstance(prefix_value, str):
raise ValueError("signed immutable publication prefix is invalid")
prefix = prefix_value.strip("/")
if not isinstance(bucket, str) or not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", bucket) or not prefix or ".." in prefix.split("/") or not isinstance(region, str) or not region:
raise ValueError("signed immutable publication location is invalid")
object_store = boto3.client(
"s3",
region_name=region,
config=Config(
connect_timeout=CONNECT_TIMEOUT,
read_timeout=READ_TIMEOUT,
retries={"max_attempts": MAXIMUM_RETRY_ATTEMPTS, "mode": "standard"},
),
)
if not isinstance(source["tables"], list) or not isinstance(removal["tables"], list) or not source["tables"] or len(source["tables"]) > maximum_tables:
raise ValueError("source/removal table population is invalid")
source_by_id: dict[str, dict[str, Any]] = {}
removal_by_id: dict[str, dict[str, Any]] = {}
for entry in source["tables"]:
if not isinstance(entry, dict) or set(entry) != TABLE_FIELDS or not isinstance(entry["table_id"], str) or SAFE.fullmatch(entry["table_id"]) is None or entry["table_id"] in source_by_id:
raise ValueError("source table entry differs or is duplicated")
source_by_id[entry["table_id"]] = entry
for entry in removal["tables"]:
if not isinstance(entry, dict) or set(entry) != REMOVAL_TABLE_FIELDS or not isinstance(entry["table_id"], str) or entry["table_id"] in removal_by_id:
raise ValueError("removal table entry differs or is duplicated")
removal_by_id[entry["table_id"]] = entry
if list(source_by_id) != sorted(source_by_id) or list(removal_by_id) != sorted(removal_by_id) or set(source_by_id) != set(removal_by_id):
raise RuntimeError("source/removal table populations differ")
args.output_directory.mkdir(parents=True, exist_ok=False)
rows: list[dict[str, Any]] = []
try:
for table_id in source_by_id:
spec = source_by_id[table_id]
removal_spec = removal_by_id[table_id]
frame = load_table(Path(spec["path"]), spec, profile)
source_ids = frame[spec["id_column"]].astype(str).tolist()
remove_ids = removal_spec["remove_ids"]
if not isinstance(remove_ids, list) or remove_ids != sorted(set(remove_ids)) or any(not isinstance(item, str) or not item for item in remove_ids) or id_digest(remove_ids) != digest(removal_spec["remove_id_population_sha256"], "removal ID population"):
raise ValueError("removal ID population differs")
if not set(remove_ids).issubset(source_ids):
raise RuntimeError("removal population includes absent source IDs")
sanitized = frame.loc[~frame[spec["id_column"]].astype(str).isin(remove_ids)].copy()
remaining_ids = sanitized[spec["id_column"]].astype(str).tolist()
expected_remaining = [item for item in source_ids if item not in set(remove_ids)]
if remaining_ids != expected_remaining or len(source_ids) - len(remaining_ids) != len(remove_ids):
raise RuntimeError("sanitized population reconciliation differs")
output = args.output_directory / (table_id + ".parquet")
with tempfile.NamedTemporaryFile(dir=args.output_directory, suffix=".parquet", delete=False) as temporary:
temporary_path = Path(temporary.name)
try:
sanitized.to_parquet(temporary_path, index=False)
os.replace(temporary_path, output)
finally:
temporary_path.unlink(missing_ok=True)
output_raw = read_file(output, positive_int(profile["maximum_source_file_bytes"], "maximum source file bytes"))
output_sha256 = hashlib.sha256(output_raw).hexdigest()
artifact_uri = publish_immutable(object_store, bucket, prefix + "/members/" + output_sha256 + ".parquet", output_raw, positive_int(profile["maximum_source_file_bytes"], "maximum source file bytes"))
rows.append({"table_id": table_id, "artifact_uri": artifact_uri, "output_sha256": output_sha256, "source_count": len(source_ids), "sanitized_count": len(remaining_ids), "removed_count": len(remove_ids), "removed_id_population_sha256": id_digest(remove_ids), "remaining_id_population_sha256": id_digest(remaining_ids)})
except Exception:
shutil.rmtree(args.output_directory, ignore_errors=True)
raise
member_population_sha256 = hashlib.sha256(canonical([{"table_id": row["table_id"], "artifact_uri": row["artifact_uri"], "output_sha256": row["output_sha256"]} for row in rows])).hexdigest()
snapshot = {"schema": "aidefend.r001002.sanitized-snapshot.v1", "tenant_id": scope[0], "incident_id": scope[1], "source_manifest_sha256": source_raw_digest, "removal_manifest_sha256": removal_raw_digest, "training_profile_sha256": profile_raw_digest, "member_population_sha256": member_population_sha256, "tables": rows}
if args.manifest_output.exists() or args.bundle_output.exists():
raise FileExistsError("sanitized manifest/bundle already exists")
manifest_bytes = canonical(snapshot) + b"\n"
output_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
output_flags |= os.O_NOFOLLOW
output_descriptor = os.open(args.manifest_output, output_flags, 0o600)
with os.fdopen(output_descriptor, "wb") as target:
target.write(manifest_bytes)
target.flush()
os.fsync(target.fileno())
subprocess.run(["cosign", "sign-blob", "--yes", "--key", profile["recovery_input_signing_key"], "--bundle", str(args.bundle_output), str(args.manifest_output)], check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT)
verified_manifest, manifest_digest = verified_json(args.manifest_output, args.bundle_output, profile["recovery_input_public_key"])
if verified_manifest != snapshot or manifest_digest != hashlib.sha256(manifest_bytes).hexdigest():
raise RuntimeError("generated sanitized manifest/bundle pair differs after signing")
manifest_uri = publish_immutable(object_store, bucket, prefix + "/manifests/" + manifest_digest + ".json", manifest_bytes, INPUT_LIMIT)
bundle_raw = read_file(args.bundle_output, INPUT_LIMIT)
bundle_digest = hashlib.sha256(bundle_raw).hexdigest()
bundle_uri = publish_immutable(object_store, bucket, prefix + "/bundles/" + bundle_digest + ".sigstore.json", bundle_raw, INPUT_LIMIT)
for field in ("training_image_digest", "training_code_digest", "initialization_policy_digest"):
digest(profile[field], field)
origin = exact_origin(profile["training_control_origin"])
if origin != exact_origin(os.environ["TRAINING_CONTROL_ORIGIN"]):
raise RuntimeError("training-control origin is not locally allowlisted")
request = {"schema": "aidefend.r001002.training-request.v1", "tenant_id": scope[0], "incident_id": scope[1], "sanitized_manifest_sha256": manifest_digest, "sanitized_manifest_uri": manifest_uri, "sanitized_manifest_bundle_sha256": bundle_digest, "sanitized_manifest_bundle_uri": bundle_uri, "sanitized_member_population_sha256": member_population_sha256, "training_image_digest": profile["training_image_digest"], "training_code_digest": profile["training_code_digest"], "initialization_policy_digest": profile["initialization_policy_digest"], "workflow_template": profile["workflow_template"]}
run = post_json(origin, "/v1/recovery-training-runs", request, scope[1] + ":" + manifest_digest)
expected_fields = {"run_id", "receipt_digest", "sanitized_manifest_sha256", "sanitized_manifest_uri", "sanitized_manifest_bundle_sha256", "sanitized_manifest_bundle_uri", "sanitized_member_population_sha256", "training_image_digest", "training_code_digest", "initialization_policy_digest", "workflow_template", "output_model_sha256"}
if set(run) != expected_fields:
raise RuntimeError("training readback receipt schema differs")
for field in ("sanitized_manifest_sha256", "sanitized_manifest_uri", "sanitized_manifest_bundle_sha256", "sanitized_manifest_bundle_uri", "sanitized_member_population_sha256", "training_image_digest", "training_code_digest", "initialization_policy_digest", "workflow_template"):
if run[field] != request[field]:
raise RuntimeError(f"training submission receipt differs: {field}")
digest(run["receipt_digest"], "training receipt")
digest(run["output_model_sha256"], "output model")
run_id = run["run_id"]
if not isinstance(run_id, str) or SAFE.fullmatch(run_id) is None:
raise RuntimeError("training run ID is invalid")
consumed = get_json(origin, "/v1/recovery-training-runs/" + quote(run_id, safe="") + "/consumed-inputs")
expected_consumed = {
"schema": "aidefend.r001002.consumed-inputs.v1",
"run_id": run_id,
"tenant_id": scope[0],
"incident_id": scope[1],
"sanitized_manifest_sha256": manifest_digest,
"sanitized_manifest_uri": manifest_uri,
"sanitized_manifest_bundle_sha256": bundle_digest,
"sanitized_manifest_bundle_uri": bundle_uri,
"sanitized_member_population_sha256": member_population_sha256,
"members": [{"table_id": row["table_id"], "artifact_uri": row["artifact_uri"], "output_sha256": row["output_sha256"]} for row in rows],
"result": "VERIFIED_CONSUMED",
}
if consumed != expected_consumed:
raise RuntimeError("independent training-control readback did not prove exact input consumption")
print(json.dumps({"outcome": "INPUT_CLOSURE_VERIFIED", "run": run, "consumption_readback_sha256": hashlib.sha256(canonical(consumed)).hexdigest(), "requires_independent_gate": "AID-R-001.002-G004"}, sort_keys=True))
if __name__ == "__main__":
main()
The sample uses versioned S3 object URIs; an equivalent store adapter is valid when it provides create-only publication, immutable version locators, byte readback, and the same manifest/member bindings. Test changed raw source, removal, or profile bytes that parse to the same JSON; duplicate keys or table/row/removal IDs; missing removal targets; extra or missing columns/tables; source overflow; local paths in the published manifest; an unpublished, wrong, changed, or partially uploaded member; create-only publication races and idempotent retry; signing failure; a sibling manifest; training receipt echo without member fetch; wrong training origin; redirect; response overflow; and independent consumption-readback mismatch. Independent replay re-materializes every signed table, compares removed and remaining populations, re-fetches every version-pinned published Parquet member plus the exact manifest/bundle pair, and confirms through the separate auditor read path that the training control plane consumed that exact closure. Submission alone is not restoration PASS; the output remains blocked pending AID-R-001.002-G004.
AID-R-001.002-G002
Use SISA shard retraining or exact federated-history replay as provisional remediation only when their architecture-specific prerequisites are present and a separate post-retraining validation gate is satisfied.
Admit only architecture-complete provisional remediation inputs
SISA is applicable only when every affected shard has signed slice lineage and a recoverable pre-deletion checkpoint. Federated replay is applicable only when the signed round population preserves every affected round, client inclusion, aggregation policy, and server-optimizer state. This code verifies immutable signed snapshots, exact sorted populations, expiry, and digest bindings before invoking one pinned workflow. It never approximates missing history or treats a workflow acknowledgement as recovery proof.
# File: remediation/run_provisional_unlearning.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from remediation.strict_io import canonical, exact_origin, post_json, read_file, verified_json
HEX = re.compile(r"^[0-9a-f]{64}$")
SAFE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
MAXIMUM_POPULATION = int(os.environ["R001_PROVISIONAL_MAXIMUM_POPULATION_COUNT"])
MAXIMUM_MEMBER_BYTES = int(os.environ["R001_PROVISIONAL_MAXIMUM_MEMBER_BYTES"])
MEMBER_KEY_ROOT = Path(os.environ["R001_PROVISIONAL_MEMBER_KEY_ROOT"]).resolve(strict=True)
if MAXIMUM_POPULATION < 1 or MAXIMUM_MEMBER_BYTES < 1 or not MEMBER_KEY_ROOT.is_dir():
raise RuntimeError("versioned provisional-remediation bounds or member-key root are invalid")
REQUEST_FIELDS = {"schema", "tenant_id", "incident_id", "method", "source_model_sha256", "method_policy_sha256", "provisional_expires_at", "training_control_origin"}
SISA_FIELDS = {"schema", "tenant_id", "incident_id", "source_model_sha256", "partition_manifest_path", "partition_manifest_sha256", "affected_shard_ids", "shard_lineage_members", "slice_population_sha256"}
SISA_MEMBER_FIELDS = {"shard_id", "checkpoint_path", "checkpoint_sha256", "lineage_path", "lineage_bundle_path", "lineage_sha256", "signer_key_id"}
LINEAGE_FIELDS = {"schema", "tenant_id", "incident_id", "source_model_sha256", "shard_id", "slice_ids", "pre_deletion_checkpoint_sha256"}
FEDERATED_FIELDS = {"schema", "tenant_id", "incident_id", "source_model_sha256", "base_checkpoint_path", "base_checkpoint_sha256", "affected_client_ids", "round_receipts", "round_receipt_population_sha256", "client_round_membership_sha256", "aggregation_policy_path", "aggregation_policy_sha256", "server_optimizer_state_path", "server_optimizer_state_sha256"}
ROUND_FIELDS = {"round_id", "receipt_path", "receipt_bundle_path", "receipt_sha256", "signer_key_id"}
ROUND_RECEIPT_FIELDS = {"schema", "tenant_id", "incident_id", "source_model_sha256", "round_id", "client_ids", "base_checkpoint_sha256", "aggregation_policy_sha256", "server_optimizer_state_sha256"}
METHODS = {"sisa_retrain": ("aidefend.sisa-lineage-receipt.v1", "sisa-retrain-affected-shards-v1"), "federated_history_replay": ("aidefend.federated-replay-inputs.v1", "federated-history-replay-v1")}
def digest(value: object, name: str) -> str:
if not isinstance(value, str) or HEX.fullmatch(value) is None or set(value) == {"0"}:
raise ValueError(f"{name} must be a nonzero SHA-256 digest")
return value
def exact_ids(value: object, name: str) -> list[str]:
if not isinstance(value, list) or not value or len(value) > MAXIMUM_POPULATION or value != sorted(set(value)) or any(not isinstance(item, str) or SAFE.fullmatch(item) is None for item in value):
raise ValueError(f"{name} must be a sorted, unique, bounded exact population")
return value
def future(value: object, name: str) -> str:
if not isinstance(value, str):
raise ValueError(f"{name} is required")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None or parsed.astimezone(timezone.utc) <= datetime.now(timezone.utc):
raise ValueError(f"{name} must be future and timezone-aware")
return value
def safe_member_path(value: object, label: str) -> Path:
if not isinstance(value, str) or not value:
raise ValueError(f"{label} path is required")
path = Path(value).resolve(strict=True)
if not path.is_file():
raise ValueError(f"{label} must be a regular staged file")
return path
def member_key(value: object) -> str:
if not isinstance(value, str) or SAFE.fullmatch(value) is None:
raise ValueError("unsafe architecture-member key ID")
path = (MEMBER_KEY_ROOT / (value + ".pub")).resolve(strict=True)
if path.parent != MEMBER_KEY_ROOT or not path.is_file():
raise PermissionError("architecture-member key is outside the fixed registry")
return str(path)
def file_digest(path_value: object, expected_value: object, label: str) -> str:
path = safe_member_path(path_value, label)
expected = digest(expected_value, label)
raw = read_file(path, MAXIMUM_MEMBER_BYTES)
if hashlib.sha256(raw).hexdigest() != expected:
raise RuntimeError(f"{label} bytes differ")
return expected
def verified_member(payload_value: object, bundle_value: object, key_id: object, expected_value: object, label: str) -> tuple[dict[str, Any], str]:
payload = safe_member_path(payload_value, label)
bundle = safe_member_path(bundle_value, label + " bundle")
if bundle != Path(str(payload) + ".sigstore.json"):
raise ValueError(f"{label} bundle must be the sibling <payload>.sigstore.json")
document, observed_digest = verified_json(payload, bundle, member_key(key_id))
if observed_digest != digest(expected_value, label):
raise RuntimeError(f"{label} signed bytes differ from the architecture receipt")
return document, observed_digest
def validate_sisa(inputs: dict[str, Any]) -> dict[str, Any]:
if set(inputs) != SISA_FIELDS:
raise ValueError("SISA input schema differs")
shards = exact_ids(inputs["affected_shard_ids"], "affected shards")
file_digest(inputs["partition_manifest_path"], inputs["partition_manifest_sha256"], "partition manifest")
members = inputs["shard_lineage_members"]
if not isinstance(members, list) or len(members) != len(shards):
raise ValueError("SISA lineage population differs")
observed_shards: list[str] = []
slice_membership: list[str] = []
lineage_receipts: list[dict[str, str]] = []
for entry in members:
if not isinstance(entry, dict) or set(entry) != SISA_MEMBER_FIELDS:
raise ValueError("SISA lineage member schema differs")
shard_id = entry["shard_id"]
if not isinstance(shard_id, str) or SAFE.fullmatch(shard_id) is None:
raise ValueError("unsafe SISA shard ID")
checkpoint_sha256 = file_digest(entry["checkpoint_path"], entry["checkpoint_sha256"], "SISA pre-deletion checkpoint")
lineage, lineage_sha256 = verified_member(entry["lineage_path"], entry["lineage_bundle_path"], entry["signer_key_id"], entry["lineage_sha256"], "SISA slice lineage")
if set(lineage) != LINEAGE_FIELDS or lineage["schema"] != "aidefend.sisa-slice-lineage.v1":
raise ValueError("SISA slice-lineage payload schema differs")
for field in ("tenant_id", "incident_id", "source_model_sha256"):
if lineage[field] != inputs[field]:
raise RuntimeError(f"SISA lineage scope differs: {field}")
if lineage["shard_id"] != shard_id or lineage["pre_deletion_checkpoint_sha256"] != checkpoint_sha256:
raise RuntimeError("SISA lineage/checkpoint binding differs")
slices = exact_ids(lineage["slice_ids"], "SISA slice IDs")
observed_shards.append(shard_id)
slice_membership.extend(shard_id + ":" + slice_id for slice_id in slices)
lineage_receipts.append({"shard_id": shard_id, "lineage_sha256": lineage_sha256, "checkpoint_sha256": checkpoint_sha256})
if observed_shards != shards or len(slice_membership) != len(set(slice_membership)):
raise RuntimeError("SISA shard or slice population is missing, duplicated, or reordered")
slice_population_sha256 = hashlib.sha256(canonical(sorted(slice_membership))).hexdigest()
if slice_population_sha256 != digest(inputs["slice_population_sha256"], "slice population"):
raise RuntimeError("SISA slice population digest differs")
return {"affected_shard_ids": shards, "slice_population_sha256": slice_population_sha256, "lineage_receipts": lineage_receipts}
def validate_federated(inputs: dict[str, Any]) -> dict[str, Any]:
if set(inputs) != FEDERATED_FIELDS:
raise ValueError("federated input schema differs")
clients = exact_ids(inputs["affected_client_ids"], "affected clients")
base_checkpoint = file_digest(inputs["base_checkpoint_path"], inputs["base_checkpoint_sha256"], "federated base checkpoint")
aggregation_policy = file_digest(inputs["aggregation_policy_path"], inputs["aggregation_policy_sha256"], "federated aggregation policy")
optimizer_state = file_digest(inputs["server_optimizer_state_path"], inputs["server_optimizer_state_sha256"], "federated server-optimizer state")
rounds = inputs["round_receipts"]
if not isinstance(rounds, list) or not rounds or len(rounds) > MAXIMUM_POPULATION:
raise ValueError("federated round population is invalid")
round_ids: list[str] = []
receipt_population: list[dict[str, str]] = []
membership: list[list[str]] = []
for entry in rounds:
if not isinstance(entry, dict) or set(entry) != ROUND_FIELDS:
raise ValueError("federated round descriptor schema differs")
round_id = entry["round_id"]
if not isinstance(round_id, str) or SAFE.fullmatch(round_id) is None:
raise ValueError("unsafe federated round ID")
receipt, receipt_sha256 = verified_member(entry["receipt_path"], entry["receipt_bundle_path"], entry["signer_key_id"], entry["receipt_sha256"], "federated round receipt")
if set(receipt) != ROUND_RECEIPT_FIELDS or receipt["schema"] != "aidefend.federated-round-receipt.v1":
raise ValueError("federated round receipt schema differs")
for field in ("tenant_id", "incident_id", "source_model_sha256"):
if receipt[field] != inputs[field]:
raise RuntimeError(f"federated round scope differs: {field}")
if receipt["round_id"] != round_id or receipt["base_checkpoint_sha256"] != base_checkpoint or receipt["aggregation_policy_sha256"] != aggregation_policy or receipt["server_optimizer_state_sha256"] != optimizer_state:
raise RuntimeError("federated round architecture-state binding differs")
round_clients = exact_ids(receipt["client_ids"], "federated round client IDs")
round_ids.append(round_id)
receipt_population.append({"round_id": round_id, "receipt_sha256": receipt_sha256})
membership.extend([[round_id, client_id] for client_id in round_clients])
if round_ids != sorted(set(round_ids)) or len(membership) != len({tuple(item) for item in membership}):
raise RuntimeError("federated round or client-to-round population is duplicated or reordered")
if sorted({item[1] for item in membership}) != clients:
raise RuntimeError("affected federated-client population differs from verified round history")
receipt_population_sha256 = hashlib.sha256(canonical(receipt_population)).hexdigest()
membership_sha256 = hashlib.sha256(canonical(sorted(membership))).hexdigest()
if receipt_population_sha256 != digest(inputs["round_receipt_population_sha256"], "round population") or membership_sha256 != digest(inputs["client_round_membership_sha256"], "client-to-round membership"):
raise RuntimeError("federated round or membership population digest differs")
return {"affected_client_ids": clients, "round_ids": round_ids, "round_receipt_population_sha256": receipt_population_sha256, "client_round_membership_sha256": membership_sha256, "base_checkpoint_sha256": base_checkpoint, "aggregation_policy_sha256": aggregation_policy, "server_optimizer_state_sha256": optimizer_state}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--request", type=Path, required=True)
parser.add_argument("--request-signature", type=Path, required=True)
parser.add_argument("--request-key", required=True)
parser.add_argument("--inputs", type=Path, required=True)
parser.add_argument("--inputs-signature", type=Path, required=True)
parser.add_argument("--inputs-key", required=True)
args = parser.parse_args()
request, request_raw_digest = verified_json(args.request, args.request_signature, args.request_key)
inputs, input_raw_digest = verified_json(args.inputs, args.inputs_signature, args.inputs_key)
if set(request) != REQUEST_FIELDS or request["schema"] != "aidefend.r001002.provisional-unlearning-request.v1" or request["method"] not in METHODS:
raise ValueError("provisional-remediation request schema differs")
method = request["method"]
expected_schema, workflow = METHODS[method]
if inputs.get("schema") != expected_schema:
raise ValueError("method prerequisite receipt schema differs")
for field in ("tenant_id", "incident_id", "source_model_sha256"):
if request[field] != inputs.get(field):
raise RuntimeError(f"request/input binding differs: {field}")
for field in ("tenant_id", "incident_id"):
if not isinstance(request[field], str) or SAFE.fullmatch(request[field]) is None:
raise ValueError(f"unsafe {field}")
digest(request["source_model_sha256"], "source model")
digest(request["method_policy_sha256"], "method policy")
expiry = future(request["provisional_expires_at"], "provisional expiry")
population = validate_sisa(inputs) if method == "sisa_retrain" else validate_federated(inputs)
origin = exact_origin(request["training_control_origin"])
if origin != exact_origin(os.environ["TRAINING_CONTROL_ORIGIN"]):
raise RuntimeError("training-control origin is not locally allowlisted")
payload = {"schema": "aidefend.r001002.provisional-unlearning-run.v1", "tenant_id": request["tenant_id"], "incident_id": request["incident_id"], "method": method, "workflow_template": workflow, "source_model_sha256": request["source_model_sha256"], "method_policy_sha256": request["method_policy_sha256"], "request_sha256": request_raw_digest, "input_receipt_sha256": input_raw_digest, "input_population": population, "provisional_expires_at": expiry}
idempotency_key = request["incident_id"] + ":" + hashlib.sha256(canonical(payload)).hexdigest()
receipt = post_json(origin, "/v1/provisional-remediation-runs", payload, idempotency_key)
fields = {"run_id", "workflow_template", "runner_image_digest", "source_model_sha256", "request_sha256", "input_receipt_sha256", "input_population", "output_model_sha256", "owner_receipt_digest", "provisional_expires_at"}
if set(receipt) != fields:
raise RuntimeError("provisional remediation receipt schema differs")
run_id = receipt["run_id"]
if not isinstance(run_id, str) or SAFE.fullmatch(run_id) is None:
raise RuntimeError("provisional remediation run_id must be a non-empty safe identifier")
for field in ("workflow_template", "source_model_sha256", "request_sha256", "input_receipt_sha256", "input_population", "provisional_expires_at"):
if receipt[field] != payload[field]:
raise RuntimeError(f"provisional remediation readback differs: {field}")
for field in ("runner_image_digest", "output_model_sha256", "owner_receipt_digest"):
digest(receipt[field], field)
print(json.dumps({"outcome": "PASS", "method": method, "run_id": run_id, "output_model_sha256": receipt["output_model_sha256"], "requires_independent_gate": "AID-R-001.002-G004"}, sort_keys=True))
if __name__ == "__main__":
main()
Exercise altered signed member bytes or bundles, an unregistered signer, missing or duplicate shard/slice/client/round membership, unavailable or wrong checkpoint bytes, incomplete slice lineage, missing client-to-round inclusion, wrong base checkpoint, aggregation policy or server-optimizer generation, cross-tenant or duplicate round receipts, wrong method schema, expired provisional window, wrong origin, redirect, response overflow, and receipt-binding mismatch. The output remains provisional: production promotion additionally requires the separate exact-artifact post-remediation validation in AID-R-001.002-G004.
AID-R-001.002-G003
For LLM incidents, run architecture-specific NPO or RMU recovery recipes only with signed forget/retain datasets and explicit provisional-release gates.
Runtime policy: Load forget/retain/attack population minima, optimization budget, sampling bounds, evaluation deadline, and release criteria from the signed, versioned unlearning plan bound to this base model and method release, bind its version/digest to run/replay evidence, and return ERROR if the plan or any value is absent, invalid, or unverifiable.
Concept:
LLM unlearning is not a generic weight-subtraction trick. Modern enterprise recovery usually separates three evidence sets: forget prompts or exemplars tied to the incident, retain prompts that must keep normal product behavior intact, and attack probes that measure whether the malicious behavior still activates. NPO-style negative-preference optimization and RMU-style representation suppression can be useful emergency recovery recipes when their architecture-specific prerequisites are present, but they remain provisional until the same independent release gate used for rollback or retraining passes.
Step 1: Bind the recipe to signed forget, retain, and attack manifests
# File: remediation/build_llm_unlearning_request.sh
#!/usr/bin/env bash
set -euo pipefail
: "${TENANT_ID:?}" "${INCIDENT_ID:?}" "${BASE_MODEL_SHA256:?}"
: "${CHECKPOINT_SHA256:?}" "${FORGET_URI:?}" "${FORGET_SHA256:?}"
: "${RETAIN_URI:?}" "${RETAIN_SHA256:?}" "${ATTACK_URI:?}" "${ATTACK_SHA256:?}"
: "${FORGET_POPULATION_SHA256:?}" "${RETAIN_POPULATION_SHA256:?}" "${ATTACK_POPULATION_SHA256:?}"
: "${FORGET_EXPECTED_COUNT:?}" "${RETAIN_EXPECTED_COUNT:?}" "${ATTACK_EXPECTED_COUNT:?}"
: "${RECIPE:?}" "${RETAIN_WEIGHT:?}" "${RETAIN_QUALITY_DROP_MAX:?}"
: "${FORBIDDEN_BEHAVIOR_RATE_MAX:?}" "${CANARY_EXTRACTION_RATE_MAX:?}"
: "${PROVISIONAL_EXPIRES_AT:?}" "${RUNNER_IMAGE_SHA256:?}" "${RUNNER_WORKFLOW_SHA256:?}"
: "${EVALUATOR_POLICY_SHA256:?}" "${EVALUATOR_RELEASE_SHA256:?}" "${RECOVERY_POLICY_KMS_KEY_URI:?}"
for digest in "$BASE_MODEL_SHA256" "$CHECKPOINT_SHA256" "$FORGET_SHA256" "$RETAIN_SHA256" "$ATTACK_SHA256" "$FORGET_POPULATION_SHA256" "$RETAIN_POPULATION_SHA256" "$ATTACK_POPULATION_SHA256" "$RUNNER_IMAGE_SHA256" "$RUNNER_WORKFLOW_SHA256" "$EVALUATOR_POLICY_SHA256" "$EVALUATOR_RELEASE_SHA256"; do
[[ "$digest" =~ ^[0-9a-f]{64}$ ]] || { echo "invalid SHA-256" >&2; exit 2; }
done
for count in "$FORGET_EXPECTED_COUNT" "$RETAIN_EXPECTED_COUNT" "$ATTACK_EXPECTED_COUNT"; do
[[ "$count" =~ ^[1-9][0-9]*$ ]] || { echo "population counts must be positive integers" >&2; exit 2; }
done
case "$RECIPE" in
npo)
: "${NPO_BETA:?}"
recipe_parameters=$(jq -en --arg beta "$NPO_BETA" --arg retain "$RETAIN_WEIGHT" '
($beta|tonumber) as $b | ($retain|tonumber) as $r |
if ($b > 0 and $r >= 0) then {beta:$b,retain_weight:$r} else error("invalid NPO parameters") end')
;;
rmu)
: "${RMU_LAYER_INDEX:?}" "${RMU_STEERING_COEFFICIENT:?}" "${RMU_CONTROL_SEED:?}"
recipe_parameters=$(jq -en --arg layer "$RMU_LAYER_INDEX" --arg steering "$RMU_STEERING_COEFFICIENT" --arg seed "$RMU_CONTROL_SEED" --arg retain "$RETAIN_WEIGHT" '
($layer|tonumber) as $l | ($steering|tonumber) as $s |
($seed|tonumber) as $seed | ($retain|tonumber) as $r |
if (($l|floor)==$l and $l>=0 and $s>0 and ($seed|floor)==$seed and $seed>=0 and $r>=0)
then {layer_index:$l,steering_coefficient:$s,control_seed:$seed,retain_weight:$r}
else error("invalid RMU parameters") end')
;;
*) echo "RECIPE must be npo or rmu" >&2; exit 2 ;;
esac
for threshold in "$RETAIN_QUALITY_DROP_MAX" "$FORBIDDEN_BEHAVIOR_RATE_MAX" "$CANARY_EXTRACTION_RATE_MAX"; do
jq -en --arg value "$threshold" '($value|tonumber) as $n | $n>=0 and $n<=1' >/dev/null || { echo "rate thresholds must be numeric values in [0,1]" >&2; exit 2; }
done
umask 077
test ! -e remediation/llm_unlearning_request.json
jq -n --arg tenant "$TENANT_ID" --arg incident "$INCIDENT_ID" --arg base "$BASE_MODEL_SHA256" --arg checkpoint "$CHECKPOINT_SHA256" --arg forget_uri "$FORGET_URI" --arg forget_sha "$FORGET_SHA256" --arg forget_pop "$FORGET_POPULATION_SHA256" --arg retain_uri "$RETAIN_URI" --arg retain_sha "$RETAIN_SHA256" --arg retain_pop "$RETAIN_POPULATION_SHA256" --arg attack_uri "$ATTACK_URI" --arg attack_sha "$ATTACK_SHA256" --arg attack_pop "$ATTACK_POPULATION_SHA256" --arg recipe "$RECIPE" --arg expires "$PROVISIONAL_EXPIRES_AT" --arg runner_image "$RUNNER_IMAGE_SHA256" --arg runner_workflow "$RUNNER_WORKFLOW_SHA256" --arg evaluator "$EVALUATOR_POLICY_SHA256" --arg evaluator_release "$EVALUATOR_RELEASE_SHA256" --argjson forget_count "$FORGET_EXPECTED_COUNT" --argjson retain_count "$RETAIN_EXPECTED_COUNT" --argjson attack_count "$ATTACK_EXPECTED_COUNT" --argjson parameters "$recipe_parameters" --argjson retain_max "$RETAIN_QUALITY_DROP_MAX" --argjson forbidden_max "$FORBIDDEN_BEHAVIOR_RATE_MAX" --argjson canary_max "$CANARY_EXTRACTION_RATE_MAX" '{
schema_version:"aidefend.llm-unlearning-request.v1",
tenant_id:$tenant, incident_id:$incident, model_family:"causal_lm",
base_model_sha256:$base, checkpoint_sha256:$checkpoint,
recipe:$recipe, release_mode:"provisional_containment",
datasets:{
forget:{uri:$forget_uri,sha256:$forget_sha,population_sha256:$forget_pop,expected_count:$forget_count},
retain:{uri:$retain_uri,sha256:$retain_sha,population_sha256:$retain_pop,expected_count:$retain_count},
attack:{uri:$attack_uri,sha256:$attack_sha,population_sha256:$attack_pop,expected_count:$attack_count}
},
recipe_parameters:$parameters,
runner:{image_sha256:$runner_image,workflow_sha256:$runner_workflow},
must_pass:{
retain_quality_drop_max:$retain_max,
forbidden_behavior_rate_max:$forbidden_max,
canary_extraction_rate_max:$canary_max
},
provisional_expires_at:$expires,
evaluator_policy_sha256:$evaluator,
evaluator_release_sha256:$evaluator_release
}' > remediation/llm_unlearning_request.json
cosign sign-blob --yes --key "$RECOVERY_POLICY_KMS_KEY_URI" --bundle remediation/llm_unlearning_request.json.sigstore.json remediation/llm_unlearning_request.json
Step 2: Run a bounded NPO-style update for harmful behavior suppression
# File: remediation/llm_npo_recovery.py
from __future__ import annotations
import torch
import torch.nn.functional as F
def sequence_logprob(model, input_ids, attention_mask, labels):
output = model(input_ids=input_ids, attention_mask=attention_mask)
logits = output.logits[:, :-1, :]
target = labels[:, 1:].clone()
valid = target.ne(-100) & attention_mask[:, 1:].bool()
token_count = valid.sum(dim=1)
if torch.any(token_count == 0):
raise ValueError("every NPO example must contain at least one scored token")
target[~valid] = 0
token_logprob = F.log_softmax(logits, dim=-1).gather(-1, target.unsqueeze(-1)).squeeze(-1)
return (token_logprob * valid).sum(dim=1) / token_count
def npo_step(
model,
reference_model,
optimizer,
forget_batch,
retain_batch,
*,
beta,
retain_weight,
):
if not torch.isfinite(torch.tensor(beta)) or beta <= 0:
raise ValueError("NPO beta must be finite and positive")
if not torch.isfinite(torch.tensor(retain_weight)) or retain_weight < 0:
raise ValueError("NPO retain_weight must be finite and non-negative")
model.train()
reference_model.eval()
optimizer.zero_grad(set_to_none=True)
forget_logprob = sequence_logprob(model, **forget_batch)
with torch.no_grad():
reference_forget_logprob = sequence_logprob(reference_model, **forget_batch)
# Penalize the recovered model when it remains as willing as the reference to emit the forbidden target.
# Zhang et al. NPO: -(2/beta) log sigmoid(-beta * log(pi_theta/pi_ref)).
npo_loss = -(2.0 / beta) * F.logsigmoid(
-beta * (forget_logprob - reference_forget_logprob)
).mean()
retain_output = model(**retain_batch)
retain_loss = retain_output.loss
loss = npo_loss + retain_weight * retain_loss
if not torch.isfinite(loss):
raise FloatingPointError("NPO produced a non-finite loss")
loss.backward()
grad_norm = torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0, error_if_nonfinite=True)
optimizer.step()
return {
"npo_loss": float(npo_loss.detach().cpu()),
"retain_loss": float(retain_loss.detach().cpu()),
"gradient_norm": float(grad_norm.detach().cpu()),
}
Step 3: Use RMU only when a target representation can be named and replayed
# File: remediation/rmu_recovery.py
import torch
def capture_hidden_state(model, batch, layer_index):
captured = {}
def hook(_module, _inputs, output):
captured["hidden"] = output[0] if isinstance(output, tuple) else output
handle = model.model.layers[layer_index].register_forward_hook(hook)
try:
model(**batch)
finally:
handle.remove()
return captured["hidden"]
def masked_mse(left, right, attention_mask):
if left.shape != right.shape:
raise ValueError("RMU representation shapes differ")
mask = attention_mask.to(device=left.device, dtype=left.dtype).unsqueeze(-1)
denominator = mask.sum() * left.shape[-1]
if denominator.item() <= 0:
raise ValueError("RMU batch contains no active tokens")
return (((left - right).pow(2)) * mask).sum() / denominator
def rmu_representation_loss(
model,
retain_model,
forget_batch,
retain_batch,
layer_index,
*,
steering_coefficient,
control_seed,
retain_weight,
):
coefficients = torch.tensor([steering_coefficient, retain_weight], dtype=torch.float64)
if not torch.isfinite(coefficients).all() or steering_coefficient <= 0 or retain_weight < 0:
raise ValueError("RMU coefficients must be finite and non-negative; steering must be positive")
if isinstance(layer_index, bool) or not isinstance(layer_index, int) or layer_index < 0:
raise ValueError("RMU layer_index must be a non-negative integer")
if isinstance(control_seed, bool) or not isinstance(control_seed, int) or control_seed < 0:
raise ValueError("RMU control_seed must be a non-negative integer")
retain_model.eval()
forget_hidden = capture_hidden_state(model, forget_batch, layer_index)
retain_hidden = capture_hidden_state(model, retain_batch, layer_index)
with torch.no_grad():
retain_reference_hidden = capture_hidden_state(retain_model, retain_batch, layer_index)
generator = torch.Generator(device=forget_hidden.device)
generator.manual_seed(control_seed)
control = torch.randn(
forget_hidden.shape[-1],
generator=generator,
device=forget_hidden.device,
dtype=forget_hidden.dtype,
)
control_norm = control.norm()
if not torch.isfinite(control_norm) or control_norm.item() <= 0:
raise FloatingPointError("RMU control vector is invalid")
control = steering_coefficient * control / control_norm
forget_loss = masked_mse(forget_hidden, control.expand_as(forget_hidden), forget_batch["attention_mask"])
retain_loss = masked_mse(retain_hidden, retain_reference_hidden, retain_batch["attention_mask"])
return forget_loss + retain_weight * retain_loss
Bind the target layer, steering coefficient, control seed, retain weight, base-model digest, and forget/retain population digests in the signed recovery policy. The deterministic seed makes the same scaled random control vector independently replayable; driving the forget representation to the zero vector is not RMU.
Step 4: Verify signed execution and evaluator receipts before issuing a provisional verdict
# File: remediation/verify_llm_recovery.py
from __future__ import annotations
import hashlib
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
REQUEST_PATH = Path(os.environ["LLM_UNLEARNING_REQUEST"])
REQUEST_BUNDLE = Path(os.environ["LLM_UNLEARNING_REQUEST_BUNDLE"])
EXECUTION_PATH = Path(os.environ["LLM_UNLEARNING_EXECUTION_RECEIPT"])
EXECUTION_BUNDLE = Path(os.environ["LLM_UNLEARNING_EXECUTION_BUNDLE"])
EVALUATION_PATH = Path(os.environ["LLM_UNLEARNING_EVALUATION_RECEIPT"])
EVALUATION_BUNDLE = Path(os.environ["LLM_UNLEARNING_EVALUATION_BUNDLE"])
OUTPUT_MODEL_PATH = Path(os.environ["LLM_UNLEARNING_OUTPUT_MODEL"])
VERDICT_PATH = Path(os.environ["LLM_UNLEARNING_VERDICT"])
VERDICT_BUNDLE = Path(os.environ["LLM_UNLEARNING_VERDICT_BUNDLE"])
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["R001_LLM_UNLEARNING_COMMAND_TIMEOUT_SECONDS"])
MAXIMUM_EVIDENCE_BYTES = int(os.environ["R001_LLM_UNLEARNING_MAXIMUM_EVIDENCE_BYTES"])
MAXIMUM_MODEL_BYTES = int(os.environ["R001_LLM_UNLEARNING_MAXIMUM_MODEL_BYTES"])
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
or not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS <= 0
or MAXIMUM_EVIDENCE_BYTES <= 0 or MAXIMUM_MODEL_BYTES <= 0):
raise RuntimeError("versioned LLM-unlearning runtime profile is invalid")
for payload, bundle, label in ((REQUEST_PATH, REQUEST_BUNDLE, "request"), (EXECUTION_PATH, EXECUTION_BUNDLE, "execution receipt"), (EVALUATION_PATH, EVALUATION_BUNDLE, "evaluation receipt"), (VERDICT_PATH, VERDICT_BUNDLE, "verdict")):
if bundle != Path(str(payload) + ".sigstore.json"):
raise RuntimeError(f"{label} bundle must be the sibling <payload>.sigstore.json")
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 verified_object(
path: Path,
bundle: Path,
public_key_env: str,
) -> tuple[dict[str, Any], bytes]:
raw = read_limited(path, MAXIMUM_EVIDENCE_BYTES)
bundle_raw = read_limited(bundle, MAXIMUM_EVIDENCE_BYTES)
if not raw or not bundle_raw:
raise ValueError("signed payload and Sigstore bundle must both be non-empty")
with tempfile.TemporaryDirectory(prefix="r001002-llm-") as directory:
snapshot_root = Path(directory)
os.chmod(snapshot_root, 0o700)
payload_snapshot = snapshot_root / "signed-evidence.json"
bundle_snapshot = snapshot_root / "signed-evidence.json.sigstore.json"
write_private_snapshot(payload_snapshot, raw)
write_private_snapshot(bundle_snapshot, bundle_raw)
subprocess.run(
["cosign", "verify-blob", "--key", os.environ[public_key_env], "--bundle", str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT_SECONDS,
)
if read_limited(payload_snapshot, MAXIMUM_EVIDENCE_BYTES) != raw:
raise RuntimeError("verified evidence payload snapshot changed")
if read_limited(bundle_snapshot, MAXIMUM_EVIDENCE_BYTES) != bundle_raw:
raise RuntimeError("verified Sigstore bundle snapshot changed")
return strict_object(raw, str(path)), raw
def sha256_file(path: Path, maximum: int) -> str:
digest = hashlib.sha256()
observed = 0
with path.open("rb") as handle:
before = os.fstat(handle.fileno())
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
observed += len(chunk)
if observed > maximum:
raise ValueError(f"{path} exceeds the versioned runtime bound")
digest.update(chunk)
after = os.fstat(handle.fileno())
if observed != before.st_size or (before.st_size, before.st_mtime_ns) != (after.st_size, after.st_mtime_ns):
raise RuntimeError(f"{path} changed during hashing")
return digest.hexdigest()
def require_sha256(value: Any, field: str) -> str:
text = str(value)
if len(text) != 64 or any(char not in "0123456789abcdef" for char in text):
raise ValueError(f"{field} must be a lowercase SHA-256 digest")
return text
def require_time(value: Any, field: str) -> datetime:
try:
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError as exc:
raise ValueError(f"{field} must be RFC 3339") from exc
if parsed.tzinfo is None:
raise ValueError(f"{field} must include a timezone")
return parsed.astimezone(timezone.utc)
def canonical_bytes(value: dict[str, Any]) -> bytes:
return (json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False) + "\n").encode("utf-8")
def write_exclusive(path: Path, payload: bytes) -> None:
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(path, flags, 0o600)
with os.fdopen(descriptor, "wb") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
request, request_bytes = verified_object(
REQUEST_PATH, REQUEST_BUNDLE, "RECOVERY_POLICY_PUBLIC_KEY"
)
execution, execution_bytes = verified_object(
EXECUTION_PATH, EXECUTION_BUNDLE, "RECOVERY_RUNNER_PUBLIC_KEY"
)
evaluation, evaluation_bytes = verified_object(
EVALUATION_PATH, EVALUATION_BUNDLE, "RECOVERY_EVALUATOR_PUBLIC_KEY"
)
if not OUTPUT_MODEL_PATH.is_file():
raise FileNotFoundError("the immutable output model artifact is unavailable")
if request.get("schema_version") != "aidefend.llm-unlearning-request.v1":
raise ValueError("unsupported request schema")
if execution.get("schema_version") != "aidefend.llm-unlearning-execution.v1":
raise ValueError("unsupported execution-receipt schema")
if evaluation.get("schema_version") != "aidefend.llm-unlearning-evaluation.v1":
raise ValueError("unsupported evaluation-receipt schema")
request_sha256 = hashlib.sha256(request_bytes).hexdigest()
output_sha256 = sha256_file(OUTPUT_MODEL_PATH, MAXIMUM_MODEL_BYTES)
require_sha256(output_sha256, "output_model_sha256")
for receipt_name, receipt in (("execution", execution), ("evaluation", evaluation)):
for field in ("tenant_id", "incident_id"):
if receipt.get(field) != request.get(field):
raise ValueError(f"{receipt_name} {field} does not match the signed request")
if receipt.get("request_sha256") != request_sha256:
raise ValueError(f"{receipt_name} is not bound to the exact signed request")
if receipt.get("output_model_sha256") != output_sha256:
raise ValueError(f"{receipt_name} is not bound to the materialized output model")
if execution.get("status") != "SUCCEEDED" or execution.get("exact_input_readback") is not True:
raise ValueError("runner did not prove successful exact-input execution")
if execution.get("recipe") != request.get("recipe"):
raise ValueError("runner used a different remediation recipe")
if execution.get("base_model_sha256") != request.get("base_model_sha256"):
raise ValueError("runner used a different base model")
for field in ("image_sha256", "workflow_sha256"):
require_sha256(request["runner"][field], f"request.runner.{field}")
if execution.get("runner", {}).get(field) != request["runner"][field]:
raise ValueError(f"runner {field} readback differs from policy")
for field in ("evaluator_policy_sha256", "evaluator_release_sha256"):
require_sha256(request[field], f"request.{field}")
if evaluation.get(field) != request[field]:
raise ValueError(f"evaluation {field} differs from policy")
for role in ("forget", "retain", "attack"):
expected = request["datasets"][role]
observed = evaluation.get("datasets", {}).get(role, {})
expected_count = expected.get("expected_count")
if not isinstance(expected_count, int) or isinstance(expected_count, bool) or expected_count <= 0:
raise ValueError(f"{role} expected_count must be a positive integer")
for field in ("sha256", "population_sha256"):
require_sha256(expected[field], f"request.datasets.{role}.{field}")
if observed.get(field) != expected[field]:
raise ValueError(f"evaluator did not read the exact {role} {field}")
if observed.get("evaluated_count") != expected_count:
raise ValueError(f"evaluator did not cover the complete {role} population")
evaluated_at = datetime.now(timezone.utc)
evaluation_observed_at = require_time(
evaluation.get("evaluated_at"), "evaluation.evaluated_at"
)
valid_until = require_time(evaluation.get("valid_until"), "evaluation.valid_until")
provisional_expires_at = require_time(request.get("provisional_expires_at"), "request.provisional_expires_at")
if (
evaluation_observed_at > evaluated_at
or evaluated_at > valid_until
or evaluated_at >= provisional_expires_at
):
raise ValueError("evaluation or provisional artifact is not currently valid")
metric_names = (
"retain_quality_drop",
"forbidden_behavior_rate",
"canary_extraction_rate",
)
metrics: dict[str, float] = {}
for name in metric_names:
value = evaluation.get("metrics", {}).get(name)
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise ValueError(f"metric {name} must be a finite number")
if value < 0 or value > 1:
raise ValueError(f"metric {name} must be in [0,1]")
metrics[name] = float(value)
limits = request["must_pass"]
expected_limits = {
"retain_quality_drop_max",
"forbidden_behavior_rate_max",
"canary_extraction_rate_max",
}
if set(limits) != expected_limits:
raise ValueError("must_pass contains missing or unknown thresholds")
for name, limit in limits.items():
if (
isinstance(limit, bool)
or not isinstance(limit, (int, float))
or not math.isfinite(limit)
or limit < 0
or limit > 1
):
raise ValueError(f"threshold {name} must be a finite number in [0,1]")
checks = {
"retain_quality_drop": metrics["retain_quality_drop"] <= float(limits["retain_quality_drop_max"]),
"forbidden_behavior_rate": metrics["forbidden_behavior_rate"] <= float(limits["forbidden_behavior_rate_max"]),
"canary_extraction_rate": metrics["canary_extraction_rate"] <= float(limits["canary_extraction_rate_max"]),
}
status = "PASS" if all(checks.values()) else "FAIL"
verdict = {
"schema_version": "aidefend.llm-unlearning-provisional-verdict.v1",
"status": status,
"tenant_id": request["tenant_id"],
"incident_id": request["incident_id"],
"request_sha256": request_sha256,
"execution_receipt_sha256": hashlib.sha256(execution_bytes).hexdigest(),
"evaluation_receipt_sha256": hashlib.sha256(evaluation_bytes).hexdigest(),
"output_model_sha256": output_sha256,
"recipe": request["recipe"],
"metrics": metrics,
"checks": checks,
"provisional_expires_at": request["provisional_expires_at"],
"requires_independent_control": "AID-R-001.002-G004",
"evaluated_at": evaluated_at.isoformat(),
}
payload = canonical_bytes(verdict)
if VERDICT_BUNDLE.exists():
raise FileExistsError("refusing to overwrite an existing verdict bundle")
write_exclusive(VERDICT_PATH, payload)
subprocess.run(
[
"cosign", "sign-blob", "--yes", "--key", os.environ["RECOVERY_VERDICT_KMS_KEY_URI"],
"--bundle", str(VERDICT_BUNDLE), str(VERDICT_PATH),
],
check=True,
capture_output=True,
text=True,
timeout=COMMAND_TIMEOUT_SECONDS,
)
verified_verdict, verified_verdict_bytes = verified_object(
VERDICT_PATH, VERDICT_BUNDLE, "RECOVERY_VERDICT_PUBLIC_KEY"
)
if verified_verdict_bytes != payload or verified_verdict != verdict:
raise RuntimeError("signed verdict snapshot differs from the generated verdict")
print(json.dumps(verdict, sort_keys=True))
if status != "PASS":
raise SystemExit(1)
Verify dependencies and result
Exercise invalid signatures, altered request or runner digests, incomplete or duplicate populations, wrong evaluator release, stale receipts, expired provisional artifacts, NaN and non-Boolean values, threshold failures, and evidence-path collisions. Store the signed request, runner receipt, evaluator receipt, model digest, provisional verdict, and Sigstore bundle in incident-scoped immutable storage. This guidance proves only that the named NPO/RMU recipe ran against the exact signed inputs and met its provisional incident-specific gate. Production promotion additionally depends on an independent AID-R-001.002-G004 PASS bound to the same output-model digest; an upstream or downstream PASS never substitutes for this control's verdict.
Action: Publish NPO/RMU recovery only as a signed, expiring provisional artifact with exact recipe parameters, dataset and population digests, runner versions, model/checkpoint digests, evaluator evidence, and an independently replayable verdict. If the incident behavior cannot be represented by defensible signed populations and replayed probes, do not claim unlearning; isolate and rollback or retrain from clean inputs.
AID-R-001.002-G004
Run a separately signed, exact-population post-remediation evaluation that checks the remediated artifact against versioned clean-utility and incident-specific attack thresholds before promotion.
Runtime policy: Load clean/attack population requirements, metric thresholds, policy issue/expiry bounds, maximum evidence age and future skew, and verifier deadline from the signed, versioned release-gate policy bound to this remediated model and evaluator release, bind its version/digest to predictions and replay, and return ERROR if the policy or any value is absent, invalid, or unverifiable.
Independent evaluation inputs
The recovery authority signs a validation policy that binds incident, tenant, the exact materialized model locator and digest, evaluator release and run, observation window, clean and attack prediction artifacts, expected sample-ID populations, metric definitions, and organization-approved thresholds. An evaluator identity outside the trainer and deployer trust domains produces the prediction artifacts and signatures. This guidance computes and signs the release verdict; it does not inherit PASS from the training job, approximate-unlearning recipe, or dataset-cleansing receipt.
Verify inputs, populations, metrics, and publish an immutable verdict
# File: remediation/validate_retraining.py
from __future__ import annotations
import hashlib
import io
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
import pandas as pd
POLICY_PATH = Path(os.environ['REMEDIATION_VALIDATION_POLICY'])
POLICY_BUNDLE = Path(os.environ['REMEDIATION_VALIDATION_POLICY_BUNDLE'])
CLEAN_PATH = Path(os.environ['CLEAN_PREDICTIONS_PATH'])
CLEAN_BUNDLE = Path(os.environ['CLEAN_PREDICTIONS_BUNDLE'])
ATTACK_PATH = Path(os.environ['ATTACK_RESULTS_PATH'])
ATTACK_BUNDLE = Path(os.environ['ATTACK_RESULTS_BUNDLE'])
MODEL_ARTIFACT_PATH = Path(os.environ['MODEL_ARTIFACT_PATH'])
OUTPUT_PATH = Path(os.environ['REMEDIATION_VERDICT_PATH'])
OUTPUT_BUNDLE = Path(os.environ['REMEDIATION_VERDICT_BUNDLE'])
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(os.environ["R001_REMEDIATION_GATE_COMMAND_TIMEOUT_SECONDS"])
MAXIMUM_EVIDENCE_BYTES = int(os.environ["R001_REMEDIATION_GATE_MAXIMUM_EVIDENCE_BYTES"])
MAXIMUM_MODEL_BYTES = int(os.environ["R001_REMEDIATION_GATE_MAXIMUM_MODEL_BYTES"])
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
or not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS <= 0
or MAXIMUM_EVIDENCE_BYTES <= 0 or MAXIMUM_MODEL_BYTES <= 0):
raise RuntimeError("versioned remediation-gate runtime profile is invalid")
for payload, bundle, label in ((POLICY_PATH, POLICY_BUNDLE, "validation policy"), (CLEAN_PATH, CLEAN_BUNDLE, "clean predictions"), (ATTACK_PATH, ATTACK_BUNDLE, "attack results"), (OUTPUT_PATH, OUTPUT_BUNDLE, "verdict")):
if bundle != Path(str(payload) + ".sigstore.json"):
raise RuntimeError(f"{label} bundle must be the sibling <payload>.sigstore.json")
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) -> bytes:
with path.open("rb") as source:
raw = source.read(MAXIMUM_EVIDENCE_BYTES + 1)
if len(raw) > MAXIMUM_EVIDENCE_BYTES:
raise ValueError(f"{path} exceeds the versioned runtime bound")
return raw
def stable_file_sha256(path: Path, maximum: int) -> str:
observed = 0
hasher = hashlib.sha256()
with path.open("rb") as source:
before = os.fstat(source.fileno())
if not os.path.isfile(path):
raise ValueError("model artifact must be a regular file")
for chunk in iter(lambda: source.read(1024 * 1024), b""):
observed += len(chunk)
if observed > maximum:
raise ValueError("model artifact exceeds the versioned runtime bound")
hasher.update(chunk)
after = os.fstat(source.fileno())
identity = lambda item: (item.st_dev, item.st_ino, item.st_size, item.st_mtime_ns, item.st_ctime_ns)
if observed != before.st_size or identity(before) != identity(after):
raise RuntimeError("model artifact changed during readback")
return hasher.hexdigest()
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_bytes(path: Path, bundle: Path, key: str) -> bytes:
raw = read_limited(path)
bundle_raw = read_limited(bundle)
if not raw or not bundle_raw:
raise ValueError("signed payload and Sigstore bundle must both be non-empty")
with tempfile.TemporaryDirectory(prefix="r001002-gate-") as directory:
snapshot_root = Path(directory)
os.chmod(snapshot_root, 0o700)
payload_snapshot = snapshot_root / "signed-input"
bundle_snapshot = snapshot_root / "signed-input.sigstore.json"
write_private_snapshot(payload_snapshot, raw)
write_private_snapshot(bundle_snapshot, bundle_raw)
subprocess.run(
['cosign', 'verify-blob', '--key', key, '--bundle', str(bundle_snapshot), str(payload_snapshot)],
check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT_SECONDS,
)
if read_limited(payload_snapshot) != raw:
raise RuntimeError("verified evidence payload snapshot changed")
if read_limited(bundle_snapshot) != bundle_raw:
raise RuntimeError("verified Sigstore bundle snapshot changed")
return raw
def require_columns(frame: pd.DataFrame, expected: set[str], label: str) -> None:
actual = set(frame.columns)
if actual != expected:
raise ValueError(f'{label} columns differ: expected={sorted(expected)}, actual={sorted(actual)}')
def exact_id_digest(values: pd.Series, label: str) -> str:
ids = [str(value) for value in values]
if any(not value for value in ids) or len(ids) != len(set(ids)):
raise ValueError(f'{label} IDs must be non-empty and unique')
canonical = json.dumps(sorted(ids), separators=(',', ':')).encode('utf-8')
return hashlib.sha256(canonical).hexdigest()
def policy_rate(value: Any, name: str) -> float:
if isinstance(value, bool):
raise ValueError(f'{name} must be numeric')
result = float(value)
if not math.isfinite(result) or not 0.0 <= result <= 1.0:
raise ValueError(f'{name} must be in [0, 1]')
return result
def strict_bool(value: Any) -> bool:
if value is True or value == 1 or value == '1' or value == 'true':
return True
if value is False or value == 0 or value == '0' or value == 'false':
return False
raise ValueError(f'invalid attack_success value: {value!r}')
policy_bytes = verified_bytes(POLICY_PATH, POLICY_BUNDLE, os.environ['RECOVERY_POLICY_PUBLIC_KEY'])
clean_bytes = verified_bytes(CLEAN_PATH, CLEAN_BUNDLE, os.environ['INDEPENDENT_EVALUATOR_PUBLIC_KEY'])
attack_bytes = verified_bytes(ATTACK_PATH, ATTACK_BUNDLE, os.environ['INDEPENDENT_EVALUATOR_PUBLIC_KEY'])
policy = strict_object(policy_bytes, "validation policy")
if policy.get('schema') != 'aidefend.r001002.validation-policy.v1':
raise ValueError('unsupported validation policy schema')
required_policy = {
'schema', 'tenant_id', 'incident_id', 'model_artifact_locator', 'model_artifact_sha256',
'evaluator_release', 'evaluator_run_id', 'observation_window_start', 'observation_window_end',
'clean_predictions_sha256', 'clean_id_population_sha256', 'clean_expected_count',
'attack_results_sha256', 'attack_id_population_sha256', 'attack_expected_count',
'clean_accuracy_min', 'attack_success_rate_max', 'policy_version',
'issued_at', 'expires_at', 'maximum_evidence_age_seconds',
'maximum_future_skew_seconds',
}
if set(policy) != required_policy:
raise ValueError('validation policy fields differ')
materialized_model_sha256 = stable_file_sha256(MODEL_ARTIFACT_PATH, MAXIMUM_MODEL_BYTES)
if materialized_model_sha256 != policy['model_artifact_sha256']:
raise ValueError('materialized model bytes differ from policy')
if str(MODEL_ARTIFACT_PATH.resolve(strict=True)) != policy['model_artifact_locator']:
raise ValueError('model artifact acquisition locator differs from policy')
if os.environ['EVALUATOR_RELEASE'] != policy['evaluator_release']:
raise ValueError('evaluator release differs from policy')
if os.environ['EVALUATOR_RUN_ID'] != policy['evaluator_run_id']:
raise ValueError('evaluator run differs from policy')
if not isinstance(policy['policy_version'], str) or not policy['policy_version'].strip():
raise ValueError('policy_version must be non-empty')
def policy_seconds(value: Any, name: str, *, allow_zero: bool = False) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f'{name} must be numeric')
result = float(value)
if not math.isfinite(result) or result < 0 or (result == 0 and not allow_zero):
raise ValueError(f'{name} is outside its permitted range')
return result
def utc(value: Any, name: str) -> datetime:
if not isinstance(value, str):
raise ValueError(f'{name} must be an RFC 3339 timestamp')
try:
parsed = datetime.fromisoformat(value.replace('Z', '+00:00'))
except ValueError as error:
raise ValueError(f'{name} must be an RFC 3339 timestamp') from error
if parsed.tzinfo is None:
raise ValueError(f'{name} must include a UTC offset')
return parsed.astimezone(timezone.utc)
evaluated_at = datetime.now(timezone.utc)
maximum_age = policy_seconds(policy['maximum_evidence_age_seconds'], 'maximum_evidence_age_seconds')
maximum_future = policy_seconds(policy['maximum_future_skew_seconds'], 'maximum_future_skew_seconds', allow_zero=True)
issued_at = utc(policy['issued_at'], 'issued_at')
expires_at = utc(policy['expires_at'], 'expires_at')
if issued_at > evaluated_at + timedelta(seconds=maximum_future):
raise ValueError('validation policy is issued too far in the future')
if expires_at <= evaluated_at or expires_at <= issued_at:
raise ValueError('validation policy is expired or has an invalid lifetime')
if (hashlib.sha256(clean_bytes).hexdigest() != policy['clean_predictions_sha256']
or hashlib.sha256(attack_bytes).hexdigest() != policy['attack_results_sha256']):
raise ValueError('signed evaluation artifact digest mismatch')
clean = pd.read_csv(io.BytesIO(clean_bytes), dtype=str, keep_default_na=False)
attack = pd.read_csv(io.BytesIO(attack_bytes), dtype=str, keep_default_na=False)
subject_columns = {'model_artifact_sha256', 'evaluator_release', 'evaluator_run_id'}
require_columns(clean, {'sample_id', 'label', 'predicted_label', 'observed_at'} | subject_columns, 'clean')
require_columns(attack, {'attack_case_id', 'attack_success', 'observed_at'} | subject_columns, 'attack')
for label, frame in (('clean', clean), ('attack', attack)):
for field in subject_columns:
values = set(frame[field])
if values != {str(policy[field])}:
raise ValueError(f'{label} evidence is not bound to the exact {field}')
clean_expected = policy['clean_expected_count']
attack_expected = policy['attack_expected_count']
if (isinstance(clean_expected, bool) or not isinstance(clean_expected, int) or clean_expected <= 0
or isinstance(attack_expected, bool) or not isinstance(attack_expected, int) or attack_expected <= 0
or len(clean) != clean_expected or len(attack) != attack_expected):
raise ValueError('evaluation population count mismatch')
if exact_id_digest(clean['sample_id'], 'clean') != policy['clean_id_population_sha256']:
raise ValueError('clean evaluation ID population mismatch')
if exact_id_digest(attack['attack_case_id'], 'attack') != policy['attack_id_population_sha256']:
raise ValueError('attack evaluation ID population mismatch')
def evidence_time(frame: pd.DataFrame, label: str) -> datetime:
values = sorted(set(frame['observed_at']))
if len(values) != 1:
raise ValueError(f'{label} must bind one common observed_at timestamp')
observed = utc(values[0], f'{label}.observed_at')
age = (evaluated_at - observed).total_seconds()
if age > maximum_age or age < -maximum_future:
raise ValueError(f'{label} evidence is stale or future-dated')
return observed
clean_observed_at = evidence_time(clean, 'clean predictions')
attack_observed_at = evidence_time(attack, 'attack results')
window_start = utc(policy['observation_window_start'], 'observation_window_start')
window_end = utc(policy['observation_window_end'], 'observation_window_end')
if window_end < window_start or any(observed < window_start or observed > window_end for observed in (clean_observed_at, attack_observed_at)):
raise ValueError('evaluation evidence falls outside the signed observation window')
clean_accuracy = float((clean['label'] == clean['predicted_label']).mean())
attack_success_rate = sum(strict_bool(value) for value in attack['attack_success']) / len(attack)
clean_min = policy_rate(policy['clean_accuracy_min'], 'clean_accuracy_min')
attack_max = policy_rate(policy['attack_success_rate_max'], 'attack_success_rate_max')
failures = []
if clean_accuracy < clean_min:
failures.append('clean_accuracy')
if attack_success_rate > attack_max:
failures.append('attack_success_rate')
verdict = {
'schema': 'aidefend.r001002.validation-verdict.v1',
'outcome': 'FAIL' if failures else 'PASS',
'tenant_id': policy['tenant_id'],
'incident_id': policy['incident_id'],
'model_artifact_locator': policy['model_artifact_locator'],
'model_artifact_sha256': materialized_model_sha256,
'evaluator_release': policy['evaluator_release'],
'evaluator_run_id': policy['evaluator_run_id'],
'observation_window_start': policy['observation_window_start'],
'observation_window_end': policy['observation_window_end'],
'policy_version': policy['policy_version'],
'policy_sha256': hashlib.sha256(policy_bytes).hexdigest(),
'runtime_profile_version': RUNTIME_PROFILE_VERSION,
'runtime_profile_sha256': RUNTIME_PROFILE_SHA256,
'evaluated_at': evaluated_at.isoformat().replace('+00:00', 'Z'),
'policy_issued_at': issued_at.isoformat().replace('+00:00', 'Z'),
'policy_expires_at': expires_at.isoformat().replace('+00:00', 'Z'),
'clean_observed_at': clean_observed_at.isoformat().replace('+00:00', 'Z'),
'attack_observed_at': attack_observed_at.isoformat().replace('+00:00', 'Z'),
'maximum_evidence_age_seconds': maximum_age,
'maximum_future_skew_seconds': maximum_future,
'clean_predictions_sha256': policy['clean_predictions_sha256'],
'attack_results_sha256': policy['attack_results_sha256'],
'clean_rows_evaluated': len(clean),
'attack_cases_evaluated': len(attack),
'clean_accuracy': clean_accuracy,
'attack_success_rate': attack_success_rate,
'thresholds': {'clean_accuracy_min': clean_min, 'attack_success_rate_max': attack_max},
'failures': failures,
}
if OUTPUT_PATH.exists() or OUTPUT_BUNDLE.exists():
raise FileExistsError('verdict output already exists')
OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
with OUTPUT_PATH.open('x', encoding='utf-8', newline='') as sink:
sink.write(json.dumps(verdict, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False) + '\n')
sink.flush()
os.fsync(sink.fileno())
subprocess.run(
['cosign', 'sign-blob', '--yes', '--key', os.environ['RECOVERY_GATE_KMS_KEY_URI'], '--bundle', str(OUTPUT_BUNDLE), str(OUTPUT_PATH)],
check=True,
capture_output=True,
text=True,
timeout=COMMAND_TIMEOUT_SECONDS,
)
verified_output = verified_bytes(OUTPUT_PATH, OUTPUT_BUNDLE, os.environ['RECOVERY_GATE_PUBLIC_KEY'])
parsed_output = strict_object(verified_output, "signed verdict")
if parsed_output != verdict:
raise RuntimeError("signed verdict snapshot differs from generated verdict")
print(json.dumps({'outcome': verdict['outcome'], 'verdict_sha256': hashlib.sha256(verified_output).hexdigest()}, sort_keys=True))
if failures:
raise SystemExit(2)
Verify the result
Exercise a caller-supplied digest that differs from the acquired file, a sibling model or evaluator run, replayed prediction rows, wrong evaluator release, changed model bytes during hashing, incomplete clean or attack IDs, and observations outside the signed window. Action: Block promotion unless the exact independently evaluated artifact receives a signed PASS verdict and an independent read identity reacquires and hashes the model, verifies the verdict and Sigstore bundle, and confirms the policy, evaluator run, observation window, and complete sample populations all name that same subject.
Open-source tools
- MLflow
- Kubeflow Pipelines
- Ray Train (distributed recovery-retraining execution)
- cleanlab (training-data issue triage before governed recovery snapshot)
- DVC
- Flower
- NVIDIA FLARE
- FATE
- PySyft
- PyTorch Geometric
- Deep Graph Library (DGL)
- GX Core
- pandas
- PyTorch
- TensorFlow
Commercial / hosted tools
- Databricks Data Intelligence Platform
- Amazon SageMaker AI
- Google Vertex AI
- Azure Machine Learning
Threat mappings
MITRE ATLAS
- AML.T0115 Publish Poisoned AI Artifacts
- AML.T0018 Manipulate AI Model
- AML.T0115.000 Publish Poisoned AI Artifacts: Datasets
- AML.T0020 Training Data Poisoning
- AML.T0059 Erode Dataset Integrity
- AML.T0031 Erode AI Model Integrity (retraining restores model integrity)
- AML.T0043.004 Craft Adversarial Data: Insert Backdoor Trigger (clean retraining removes the compromised model association so the registered trigger no longer activates the incident behavior)
- AML.T0018.000 Manipulate AI Model: Poison AI Model
MAESTRO
- Backdoor Attacks (L1)
- Data Poisoning (Training Phase) (L1)
- Data Poisoning (L2)
OWASP LLM Top 10 2026
- LLM05:2026 Data and Model Poisoning
OWASP ML Top 10 2023
- ML02:2023 Data Poisoning Attack
- ML10:2023 Model Poisoning
- ML07:2023 Transfer Learning Attack (retraining on verified base removes transfer attack vectors)
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities
NIST Adversarial Machine Learning 2025
- NISTAML.012 Clean-label Poisoning
- NISTAML.013 Data Poisoning
- NISTAML.021 Clean-label Backdoor
- NISTAML.023 Backdoor Poisoning
- NISTAML.024 Targeted Poisoning
- NISTAML.026 Model Poisoning (Integrity)
- NISTAML.011 Model Poisoning (Availability) (retraining restores model availability)
- NISTAML.051 Model Poisoning (Supply Chain) (retraining from verified sources remediates supply chain poisoning)
- NISTAML.037 Training Data Attacks
Cisco Integrated AI Security and Safety Framework
- AISubtech-10.1.3 Sensitive Data Reconstruction (Suppresses the incident-defined training-canary extraction behavior measured after recovery)
- AISubtech-8.2.1 Training Data Exposure (Suppresses the incident-defined training-canary extraction behavior measured after recovery)
- AISubtech-9.2.2 Backdoors and Trojans (Remediates the measured model backdoor behavior; original-artifact eviction is separate)
- AITech-15.1 Harmful Content (Suppresses only the incident-defined harmful behavior represented in the signed recovery probes)
- AITech-6.1 Training Data Poisoning
- AITech-8.2 Data Exfiltration / Exposure (Suppresses the incident-defined training-canary extraction behavior measured after recovery)
Google Secure AI Framework 2.0 - Risks
- DP: Data Poisoning (retraining on sanitized data eliminates poisoning effects)
- UTD: Unauthorized Training Data (retraining ensures only authorized data is used)
- MST: Model Source Tampering (fresh retraining from verified code eliminates source tampering)
Databricks AI Security Framework 3.0
- Datasets 3.1: Data poisoning (retraining on clean data eliminates poisoned samples)
- Datasets 3.3: Label flipping (retraining removes label-flipped data from training process)
- Model 7.1: Backdoor machine learning / Trojaned model (retraining from scratch removes embedded backdoors)
- Raw Data 1.7: Lack of data trustworthiness (retraining on verified data restores trustworthiness)