Scope boundary

Owns the signed reconstruction model, governed clean media population, reconstruction-error reference statistics, controlled recalibration, and independent replay evidence for an approved image or media domain.

AID-M-003.002: AID-M-003.005 owns a modality-specific reconstruction-error reference; AID-M-003.002 owns general predictive and operational baselines. Reconstruction behavior is a specialized media-domain reference, not a generic accuracy, latency, or output-distribution metric.

AID-D-001.002: AID-M-003.005 creates the trusted reconstruction-error reference; AID-D-001.002 scores runtime media, emits forensic findings, and may cluster suspicious bursts. Reference generation does not classify an ingress artifact, while the detector must consume rather than recalibrate the approved reference.

Implementation guidance

• AID-M-003.005-G001: Generate and independently replay an expiring signed reconstruction-error reference bound to the exact reconstruction pipeline, model, preprocessing, policy, and governed clean holdout population.
• AID-M-003.005-G002: Recalibrate only after an approved reconstruction or clean-domain change; renew validity separately, and independently replay, promote, and roll back the model, preprocessing, and reference as one complete release unit.

AID-M-003.005-G001

Use one reconstruction pipeline at build and ingress

Apply this control when an approved media domain uses reconstruction error to flag anomalies. This runnable adapter covers single-frame PNG/JPEG images with the same flat autoencoder, RGB preprocessing, and per-image mean squared error at build time and in AID-D-001.002-G005. Save the shared module below beside both entry points and pin its bytes, Python dependencies, CPU type and numerical libraries in the deployment image. Other media or an inverter/generator architecture need a reviewed adapter on both ends: measure its actual reconstruction pipeline and error definition; do not rename fields or invent inverter/generator identities to reuse unrelated statistics. An unsupported shape or format in an applicable domain is an implementation error, not NOT_APPLICABLE.

Bind immutable inputs and a policy-defined validity window

Keep the reviewed clean holdout separate from training data. Its manifest contains exactly {schema_version: "aidefend.clean-media-manifest.v1", items: [{path, sha256}]}, with absolute paths and the complete approved population. The signed policy supplies an independently declared expected_samples, a minimum, quantile, domain, reference version, absolute artifact paths and digests, and UTC issued_at/expires_at within maximum_reference_age_seconds. Fix these release metadata before independent replay; the builder never inserts its current clock into output bytes. Resolve all limits and validity periods from policy rather than this example.

The preprocessing JSON uses aidefend.rgb-preprocessing.v1, color_mode=RGB, resize=BILINEAR, layout=CHW, dtype=float32, value_scale=uint8/255, and policy-supplied image dimensions, byte/pixel bounds and accepted_formats (PNG/JPEG). The model config contains input_dim, hidden_dim, latent_dim; input_dim must equal three times the resized pixel count. The shared code applies that exact preprocessing to each digest-bound raw holdout image and each ingress image. Review the clean population; neither its manifest nor a successful signature proves it is clean.

Save the shared, pinned implementation

from __future__ import annotations

# File: reconstruction_runtime.py -- the same pinned module at build and ingress.
import hashlib
import io
import json
import math
import os
import re
import stat
import subprocess
import tempfile
import warnings
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

REFERENCE_SCHEMA = "aidefend.reconstruction-reference.v2"
POLICY_FIELDS = {
    "schema_version", "policy_version", "reference_version", "media_domain",
    "pipeline_sha256", "model_config_path", "model_config_sha256",
    "model_weights_path", "model_weights_sha256", "preprocessing_path",
    "preprocessing_sha256", "clean_holdout_path", "clean_holdout_sha256",
    "expected_samples", "minimum_samples", "baseline_quantile",
    "issued_at", "expires_at", "maximum_reference_age_seconds",
}
REFERENCE_FIELDS = {
    "schema_version", "status", "reference_version", "policy_version",
    "policy_sha256", "media_domain", "pipeline_sha256", "model_config_sha256",
    "model_weights_sha256", "preprocessing_sha256", "clean_holdout_sha256",
    "expected_samples", "measured_samples", "issued_at", "expires_at",
    "error_metric", "mean_mse", "std_mse", "baseline_quantile",
    "quantile_mse", "quantile_method",
}


def sha256(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


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


def strict_json(raw: bytes) -> dict:
    def unique(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate JSON key: " + key)
            result[key] = value
        return result

    def reject(value):
        raise ValueError("non-finite JSON number: " + value)

    result = json.loads(raw.decode("utf-8", errors="strict"),
                        object_pairs_hook=unique, parse_constant=reject)
    if not isinstance(result, dict):
        raise ValueError("expected a JSON object")
    return result


def positive_integer(value, label):
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(label + " must be a positive policy integer")
    return value


def finite_number(value, label):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(label + " must be numeric")
    result = float(value)
    if not math.isfinite(result):
        raise ValueError(label + " must be finite")
    return result


def timestamp(value):
    if not isinstance(value, str) or not re.fullmatch(
        r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", value
    ):
        raise ValueError("expected a UTC timestamp with seconds and Z")
    return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)


def validate_policy(policy):
    if set(policy) != POLICY_FIELDS or policy["schema_version"] != "aidefend.reconstruction-policy.v2":
        raise ValueError("reconstruction policy schema mismatch")
    for key in ("policy_version", "reference_version", "media_domain"):
        if not isinstance(policy[key], str) or not policy[key].strip():
            raise ValueError("missing policy identity: " + key)
    for key in ("pipeline_sha256", "model_config_sha256", "model_weights_sha256",
                "preprocessing_sha256", "clean_holdout_sha256"):
        if not isinstance(policy[key], str) or not re.fullmatch(r"[a-f0-9]{64}", policy[key]):
            raise ValueError("invalid signed digest: " + key)
    for key in ("model_config_path", "model_weights_path", "preprocessing_path", "clean_holdout_path"):
        if not isinstance(policy[key], str) or not Path(policy[key]).is_absolute():
            raise ValueError("artifact path must be absolute: " + key)
    count = positive_integer(policy["expected_samples"], "expected_samples")
    if count < positive_integer(policy["minimum_samples"], "minimum_samples"):
        raise ValueError("INSUFFICIENT_DATA: holdout is below the policy minimum")
    quantile = finite_number(policy["baseline_quantile"], "baseline_quantile")
    if not 0 < quantile < 1:
        raise ValueError("baseline_quantile must be inside (0,1)")
    maximum_age = positive_integer(policy["maximum_reference_age_seconds"], "maximum_reference_age_seconds")
    age = (timestamp(policy["expires_at"]) - timestamp(policy["issued_at"])).total_seconds()
    if not 0 < age <= maximum_age:
        raise ValueError("reference validity exceeds the signed policy window")


def require_current(policy, now=None):
    # Only a trusted service clock supplies now, never a request parameter.
    now = datetime.now(timezone.utc) if now is None else now
    if now.tzinfo is None or not timestamp(policy["issued_at"]) <= now < timestamp(policy["expires_at"]):
        raise ValueError("INSUFFICIENT_DATA: reference is not currently valid")


def regular_bytes(path: Path, maximum_bytes: int) -> bytes:
    positive_integer(maximum_bytes, "maximum_bytes")
    before = path.lstat()
    if not stat.S_ISREG(before.st_mode):
        raise ValueError("non-regular or symlink artifact")
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_BINARY", 0)
                         | getattr(os, "O_NOFOLLOW", 0))
    with os.fdopen(descriptor, "rb") as handle:
        opened = os.fstat(handle.fileno())
        if (not stat.S_ISREG(opened.st_mode)
                or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino)
                or not 0 < opened.st_size <= maximum_bytes):
            raise ValueError("artifact changed or exceeds its byte bound")
        raw = handle.read(maximum_bytes + 1)
    if not raw or len(raw) > maximum_bytes:
        raise ValueError("empty or oversized artifact")
    return raw


def bound_bytes(path, expected_digest, maximum_bytes):
    raw = regular_bytes(Path(path), maximum_bytes)
    if sha256(raw) != expected_digest:
        raise ValueError("artifact differs from the signed deployment binding")
    return raw


def verified_bytes(path, signature, verify_key, expected_digest,
                   *, maximum_bytes, command_timeout):
    # Paths, key, digest and bounds come from the protected deployment profile.
    timeout = finite_number(command_timeout, "command_timeout")
    if timeout <= 0:
        raise ValueError("command_timeout must be positive")
    raw = bound_bytes(path, expected_digest, maximum_bytes)
    signature_raw = regular_bytes(Path(signature), maximum_bytes)
    with tempfile.TemporaryDirectory(prefix="aidefend-reference-verify-") as directory:
        snapshot = Path(directory) / "artifact.json"
        bundle = Path(directory) / "artifact.sig"
        snapshot.write_bytes(raw)
        bundle.write_bytes(signature_raw)
        subprocess.run(["cosign", "verify-blob", "--key", str(verify_key),
                        "--bundle", str(bundle), str(snapshot)], check=True,
                       capture_output=True, timeout=timeout)
        if snapshot.read_bytes() != raw or bundle.read_bytes() != signature_raw:
            raise ValueError("verification snapshot changed")
    return raw


def validate_preprocessing(value):
    expected = {"schema_version", "image_width", "image_height", "maximum_file_bytes",
                "maximum_pixels", "accepted_formats", "color_mode", "resize",
                "layout", "dtype", "value_scale"}
    if set(value) != expected or value["schema_version"] != "aidefend.rgb-preprocessing.v1":
        raise ValueError("preprocessing contract schema mismatch")
    for key in ("image_width", "image_height", "maximum_file_bytes", "maximum_pixels"):
        positive_integer(value[key], key)
    if value["image_width"] * value["image_height"] > value["maximum_pixels"]:
        raise ValueError("resized image exceeds the pixel bound")
    if any(value[key] != expected_value for key, expected_value in {
        "color_mode": "RGB", "resize": "BILINEAR", "layout": "CHW",
        "dtype": "float32", "value_scale": "uint8/255",
    }.items()):
        raise ValueError("unsupported preprocessing operation")
    formats = value["accepted_formats"]
    if (not isinstance(formats, list) or not formats or len(set(formats)) != len(formats)
            or any(x not in ("PNG", "JPEG") for x in formats)):
        raise ValueError("unsupported image format in this adapter")


def preprocess_image(raw, preprocessing):
    import numpy as np
    import torch
    from PIL import Image

    validate_preprocessing(preprocessing)
    if not raw or len(raw) > preprocessing["maximum_file_bytes"]:
        raise ValueError("image exceeds the signed byte bound")
    with warnings.catch_warnings():
        warnings.simplefilter("error", Image.DecompressionBombWarning)
        with Image.open(io.BytesIO(raw)) as probe:
            if probe.format not in preprocessing["accepted_formats"] or getattr(probe, "n_frames", 1) != 1:
                raise ValueError("unsupported image format or multi-frame image")
            width, height = probe.size
            if width < 1 or height < 1 or width * height > preprocessing["maximum_pixels"]:
                raise ValueError("image exceeds the signed pixel bound")
            probe.verify()
        with Image.open(io.BytesIO(raw)) as image:
            image.load()
            resized = image.convert("RGB").resize(
                (preprocessing["image_width"], preprocessing["image_height"]),
                resample=Image.Resampling.BILINEAR,
            )
            values = np.asarray(resized, dtype=np.float32).transpose(2, 0, 1).copy() / 255.0
    return torch.from_numpy(values).reshape(1, -1)


def make_model(config):
    from torch import nn

    if set(config) != {"input_dim", "hidden_dim", "latent_dim"}:
        raise ValueError("model configuration schema mismatch")
    for key, value in config.items():
        positive_integer(value, key)

    class Autoencoder(nn.Module):
        def __init__(self):
            super().__init__()
            self.encoder = nn.Sequential(nn.Linear(config["input_dim"], config["hidden_dim"]),
                                         nn.ReLU(), nn.Linear(config["hidden_dim"], config["latent_dim"]))
            self.decoder = nn.Sequential(nn.Linear(config["latent_dim"], config["hidden_dim"]),
                                         nn.ReLU(), nn.Linear(config["hidden_dim"], config["input_dim"]))

        def forward(self, batch):
            return self.decoder(self.encoder(batch))

    return Autoencoder()


@dataclass(frozen=True)
class Runtime:
    policy_raw: bytes
    policy: dict
    preprocessing: dict
    model: object
    maximum_artifact_bytes: int


def load_runtime(policy_raw, *, maximum_artifact_bytes):
    # policy_raw must be captured by verified_bytes, not supplied by the client.
    import torch

    policy = strict_json(policy_raw)
    validate_policy(policy)
    bound_bytes(__file__, policy["pipeline_sha256"], maximum_artifact_bytes)
    config_raw = bound_bytes(policy["model_config_path"], policy["model_config_sha256"], maximum_artifact_bytes)
    weights_raw = bound_bytes(policy["model_weights_path"], policy["model_weights_sha256"], maximum_artifact_bytes)
    preprocessing_raw = bound_bytes(policy["preprocessing_path"], policy["preprocessing_sha256"], maximum_artifact_bytes)
    config, preprocessing = strict_json(config_raw), strict_json(preprocessing_raw)
    validate_preprocessing(preprocessing)
    model = make_model(config)
    if config["input_dim"] != 3 * preprocessing["image_width"] * preprocessing["image_height"]:
        raise ValueError("model shape differs from preprocessing")
    model.load_state_dict(torch.load(io.BytesIO(weights_raw), map_location="cpu", weights_only=True), strict=True)
    model.eval()
    return Runtime(policy_raw, policy, preprocessing, model, maximum_artifact_bytes)


def reconstruction_errors(runtime, batch):
    import torch

    if (batch.ndim != 2 or batch.shape[1] != 3 * runtime.preprocessing["image_width"] * runtime.preprocessing["image_height"]
            or not torch.isfinite(batch).all()):
        raise ValueError("preprocessed tensor is invalid")
    with torch.inference_mode():
        reconstructed = runtime.model(batch)
        if reconstructed.shape != batch.shape or not torch.isfinite(reconstructed).all():
            raise ValueError("model reconstruction is invalid")
        errors = torch.mean((batch - reconstructed) ** 2, dim=1)
    if not torch.isfinite(errors).all():
        raise ValueError("reconstruction error is non-finite")
    return errors.cpu().numpy()


def reference_bindings(runtime):
    policy = runtime.policy
    return {**{key: policy[key] for key in (
        "reference_version", "policy_version", "media_domain", "pipeline_sha256",
        "model_config_sha256", "model_weights_sha256", "preprocessing_sha256",
        "clean_holdout_sha256", "expected_samples", "issued_at", "expires_at",
    )}, "policy_sha256": sha256(runtime.policy_raw)}


def build_reference(runtime, *, now=None):
    import numpy as np

    require_current(runtime.policy, now)
    raw = bound_bytes(runtime.policy["clean_holdout_path"], runtime.policy["clean_holdout_sha256"],
                      runtime.maximum_artifact_bytes)
    manifest = strict_json(raw)
    if set(manifest) != {"schema_version", "items"} or manifest["schema_version"] != "aidefend.clean-media-manifest.v1":
        raise ValueError("holdout manifest schema mismatch")
    items = manifest["items"]
    if not isinstance(items, list) or len(items) != runtime.policy["expected_samples"]:
        raise ValueError("INSUFFICIENT_DATA: holdout population is incomplete")
    errors, seen = [], set()
    for item in items:
        if (not isinstance(item, dict) or set(item) != {"path", "sha256"}
                or not isinstance(item["path"], str) or not Path(item["path"]).is_absolute()
                or item["path"] in seen):
            raise ValueError("invalid or repeated holdout member")
        seen.add(item["path"])
        image = bound_bytes(item["path"], item["sha256"], runtime.preprocessing["maximum_file_bytes"])
        errors.extend(reconstruction_errors(runtime, preprocess_image(image, runtime.preprocessing)).tolist())
    scores = np.asarray(errors, dtype=np.float64)
    result = {"schema_version": REFERENCE_SCHEMA, "status": "PASS", **reference_bindings(runtime),
              "measured_samples": len(errors), "error_metric": "mean_squared_error.float32.per_image",
              "mean_mse": float(np.mean(scores)), "std_mse": float(np.std(scores, ddof=0)),
              "baseline_quantile": float(runtime.policy["baseline_quantile"]),
              "quantile_mse": float(np.quantile(scores, runtime.policy["baseline_quantile"], method="linear")),
              "quantile_method": "numpy.linear"}
    validate_reference(runtime, result, now=now)
    return result


def validate_reference(runtime, value, *, now=None):
    validate_policy(runtime.policy)
    require_current(runtime.policy, now)
    if set(value) != REFERENCE_FIELDS or value["schema_version"] != REFERENCE_SCHEMA or value["status"] != "PASS":
        raise ValueError("verified reference schema mismatch")
    if any(value[key] != expected for key, expected in reference_bindings(runtime).items()):
        raise ValueError("reference is bound to another release or validity window")
    if (positive_integer(value["expected_samples"], "expected_samples") != runtime.policy["expected_samples"]
            or positive_integer(value["measured_samples"], "measured_samples") != runtime.policy["expected_samples"]
            or value["error_metric"] != "mean_squared_error.float32.per_image"
            or value["quantile_method"] != "numpy.linear"
            or finite_number(value["baseline_quantile"], "baseline_quantile") != runtime.policy["baseline_quantile"]):
        raise ValueError("reference population or error definition differs")
    for key in ("mean_mse", "std_mse", "quantile_mse"):
        if finite_number(value[key], key) < 0:
            raise ValueError("negative reference statistic: " + key)

Build and sign an immutable candidate

Install PyTorch, NumPy, Pillow and cosign in the version-pinned build image. Inject the required bounds, protected policy path/signature/key/digest, candidate root and signing/verification keys named below from the trusted deployment profile. The policy signature is verified over captured bytes before parsing; model, preprocessing and pipeline bytes must match it. The signing identity must be separate from the read-only replay and runtime identities. Run python build_reconstruction_reference.py; it prints a fresh candidate directory but never activates it.

from __future__ import annotations

# File: build_reconstruction_reference.py; put beside reconstruction_runtime.py.
import os
import hashlib
import subprocess
import tempfile
from pathlib import Path

from reconstruction_runtime import canonical, load_runtime, regular_bytes, verified_bytes, build_reference


def main():
    # Required values are injected by the verified deployment profile, not the caller.
    maximum_bytes = int(os.environ["M003005_ARTIFACT_MAX_BYTES"])
    timeout = float(os.environ["M003005_RECONSTRUCTION_COMMAND_TIMEOUT_SECONDS"])
    policy_raw = verified_bytes(
        os.environ["RECONSTRUCTION_POLICY_PATH"], os.environ["RECONSTRUCTION_POLICY_SIGNATURE"],
        os.environ["RECONSTRUCTION_POLICY_VERIFY_KEY"], os.environ["RECONSTRUCTION_POLICY_SHA256"],
        maximum_bytes=maximum_bytes, command_timeout=timeout,
    )
    runtime = load_runtime(policy_raw, maximum_artifact_bytes=maximum_bytes)
    payload = canonical(build_reference(runtime))
    # A fresh immutable directory is the candidate unit; never overwrite an active pair.
    root = Path(os.environ["RECONSTRUCTION_CANDIDATE_ROOT"]).resolve(strict=True)
    directory = Path(tempfile.mkdtemp(prefix="reference-", dir=root))
    artifact, signature = directory / "reference.json", directory / "reference.sig"
    artifact.write_bytes(payload)
    subprocess.run(["cosign", "sign-blob", "--yes", "--key", "env://RECONSTRUCTION_BASELINE_SIGNING_KEY",
                    "--bundle", str(signature), str(artifact)], check=True, timeout=timeout)
    verified_bytes(artifact, signature, os.environ["RECONSTRUCTION_BASELINE_VERIFY_KEY"],
                   hashlib.sha256(payload).hexdigest(), maximum_bytes=maximum_bytes,
                   command_timeout=timeout)
    if regular_bytes(artifact, maximum_bytes) != payload:
        raise ValueError("candidate changed during signing")
    print(directory)
    # A separate read-only replay and G002 promotion are required before activation.


if __name__ == "__main__":
    main()

Independent replay and downstream acceptance

A separate read-only identity verifies the policy and candidate signature/digest, loads the same immutable artifacts and entire holdout, and compares canonical(build_reference(runtime)) byte for byte with the signed candidate. Then call the actual AID-D-001.002-G005 scorer with that candidate and clean/anomalous regression images. The scorer consumes aidefend.reconstruction-reference.v2 directly; legacy v1 artifacts require a measured rebuild, not a schema-only conversion. Missing members, unreadable media, non-finite results or replay differences block promotion and remain INSUFFICIENT_DATA or ERROR, never a valid reference. Runtime anomaly thresholds remain Detect policy, not baseline recalibration.

AID-M-003.005-G002

Recalibrate only for an approved measurement change

Create a new reference when the reconstruction model/configuration, shared pipeline code, preprocessing contract, applicable media domain, or governed clean holdout population changes. The change record binds the prior reference digest, new model, pipeline, preprocessing and holdout digests, policy version, reason, approver and intended activation. Never update the active distribution from unreviewed production inputs.

Renew validity without silently changing the distribution

Expiry is not itself a reason to retrain, recalibrate or move a threshold. Before renewal, confirm that the approved domain and all measurement bindings remain suitable and unchanged. Issue a new signed policy/reference version with a finite policy-approved validity window, independently replay the same immutable inputs, and require identical statistics. Preserve the old signed artifact; never edit its expiry in place. Changed inputs or an unexplained statistical shift require the approved recalibration procedure. While no currently valid reference is available, runtime scoring is INSUFFICIENT_DATA, not a clean result.

Replay and promote one complete release unit

Run the same pinned builder, retain old and candidate statistics, and execute the approved clean and anomaly suites against both. An independent read-only identity verifies captured signatures/digests and requires byte-identical canonical replay, including the predeclared validity metadata. Promote the reconstruction model/configuration, shared pipeline, preprocessing artifact, policy, holdout digest, reference artifact and signature through one protected atomic release pointer. Do not overwrite its members individually. Retain that complete previous unit for rollback; rollback must restore preprocessing as well as model and reference, and must not revive an expired reference. Pin image/dependencies/processor for exact replay; any numerical-tolerance adapter is an explicitly reviewed alternative, not this byte-identical method.

Open-source tools

• PyTorch
• TensorFlow
• Keras
• NumPy
• OpenCV
• Pillow
• Anomalib
• MLflow
• Apache Airflow
• Kubeflow Pipelines
• Cosign

Commercial / hosted tools

• Amazon SageMaker AI
• Google Vertex AI
• Azure Machine Learning
• Databricks Machine Learning
• Weights & Biases

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• N/A

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• N/A

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Evaluation 6.2: Insufficient evaluation data