AID-H-013.001: Digital Content Watermarking & Signed Provenance
Embed and verify policy-bound provenance signals on supported production release paths: VideoSeal/PixelSeal for image releases, the maintained Transformers SynthID Text generation/detector path for compatible autoregressive text systems, and C2PA Content Credentials for supported signed image, audio, or video containers. Each path binds exact artifacts, calibration or trust material, a complete policy-selected test population, and independently replayable evidence.
Pillars: data, model
Lifecycle: building
Scope boundary
Owns supported production paths for digital-content watermark embedding and verification and C2PA-style signed provenance, with exact artifact binding, calibration, trust material, complete test populations, and replayable evidence.
AID-H-013.002: AID-H-013.001 places or verifies provenance signals in generated content; AID-H-013.002 compares complete authorized suspect model bytes with a signed release fingerprint. Content provenance and model-artifact ownership evidence apply to different assets.
AID-DV-004: AID-H-013.001 emits production provenance or watermark signals on supported release paths; AID-DV-004 plants deception markers, trap prompts, and leak beacons intended to trigger on misuse. A general watermark is not a deception trap, and a trap marker is not universal content provenance.
Warning
Watermark Robustness Is Limited
Invisible watermarks and text-watermark detectors are provenance and misuse-tracing signals, not proof that an asset is authentic or impossible to alter. They can be stripped, weakened, copied, paraphrased, translated, or forged under known attack patterns, including watermark-copying and diffusion-forging research. C2PA / Content Credentials add cryptographic provenance for supported media containers, but they also depend on signer trust, platform preservation, and verifier policy. Pair watermarking with signed provenance, controlled key management, release evidence, and forensic review for high-impact claims.
Implementation guidance
- AID-H-013.001-G001: Embed policy-bound image watermarks with a pinned VideoSeal or PixelSeal model and independently replay robustness and payload recovery before release.
- AID-H-013.001-G002: Apply the maintained Hugging Face SynthID Text generation path and a policy-bound detector whose calibration and artifacts are independently verified.
- AID-H-013.001-G003: Attach C2PA Content Credentials with a managed signer and admit them only after pinned trust-list verification and independent tamper replay.
AID-H-013.001-G001
Embed policy-bound image watermarks with a pinned VideoSeal or PixelSeal model and independently replay robustness and payload recovery before release.
Use this control for JPEG, PNG, or WebP image release paths that can run a pinned VideoSeal or PixelSeal model before distribution and can retain the original asset, watermark payload, and detector calibration. Video watermarking requires a separate frame/audio/container pipeline and is not claimed by this implementation. The signed policy owns the model name, exact loaded-state digest, package version, embedding strength, required transformation population, and detector thresholds; none of those values is a universal AIDEFEND default. Run the worker in an immutable, network-denied image with a pre-populated model cache.
Runnable image embedding and independent detector gate
# File: watermarking/videoseal_release_gate.py
from __future__ import annotations
import math
import hashlib
import hmac
import io
import json
import os
import subprocess
import tempfile
from importlib.metadata import version
from pathlib import Path
import torch
import torchvision.transforms.functional as TF
import videoseal
from PIL import Image
def _required_positive_seconds(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number of seconds")
return value
IMAGE_WATERMARK_POLICY_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_IMAGE_WATERMARK_POLICY_VERIFY_TIMEOUT_SECONDS")
class InsufficientData(RuntimeError):
pass
def sha256_bytes(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def verified_policy(path: Path, bundle: Path, identity: str, issuer: str) -> tuple[dict, str]:
raw = path.read_bytes()
bundle_raw = bundle.read_bytes()
if not raw or not bundle_raw:
raise ValueError("policy or verification bundle is empty")
with tempfile.TemporaryDirectory(prefix="watermark-policy-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
snapshot = root / "policy.json"
bundle_snapshot = root / "bundle.json"
snapshot.write_bytes(raw)
bundle_snapshot.write_bytes(bundle_raw)
os.chmod(snapshot, 0o400)
os.chmod(bundle_snapshot, 0o400)
subprocess.run([
"cosign", "verify-blob", "--bundle", str(bundle_snapshot),
"--certificate-identity", identity,
"--certificate-oidc-issuer", issuer, str(snapshot),
], check=True, capture_output=True, text=True, timeout=IMAGE_WATERMARK_POLICY_VERIFY_TIMEOUT_SECONDS)
if not hmac.compare_digest(snapshot.read_bytes(), raw):
raise ValueError("verified policy snapshot changed")
policy = json.loads(raw.decode("utf-8", errors="strict"))
required = {
"schema_version", "policy_version", "model_name",
"videoseal_package_version", "loaded_state_sha256", "scaling_w",
"minimum_presence_logit", "minimum_bit_accuracy",
"required_transforms",
}
if not isinstance(policy, dict) or set(policy) != required:
raise ValueError("watermark policy schema differs")
if policy["schema_version"] != "aidefend.media-watermark-policy.v1":
raise ValueError("unsupported watermark policy")
if policy["model_name"] not in {"videoseal", "pixelseal"}:
raise ValueError("unsupported production watermark model")
if version("videoseal") != policy["videoseal_package_version"]:
raise ValueError("videoseal package version differs")
if len(policy["loaded_state_sha256"]) != 64:
raise ValueError("loaded-state digest is invalid")
int(policy["loaded_state_sha256"], 16)
transforms = policy["required_transforms"]
if not isinstance(transforms, list) or not transforms:
raise InsufficientData("required transformation population is empty")
return policy, sha256_bytes(raw)
def loaded_state_sha256(model: torch.nn.Module) -> str:
digest = hashlib.sha256()
for name, tensor in sorted(model.state_dict().items()):
value = tensor.detach().cpu().contiguous()
header = json.dumps(
{"name": name, "dtype": str(value.dtype), "shape": list(value.shape)},
sort_keys=True, separators=(",", ":"),
).encode()
digest.update(header)
digest.update(value.numpy().tobytes(order="C"))
return digest.hexdigest()
def transform_image(image: torch.Tensor, spec: dict) -> torch.Tensor:
if not isinstance(spec, dict) or set(spec) != {"id", "kind", "value"}:
raise ValueError("transformation schema differs")
kind = spec["kind"]
value = spec["value"]
if kind == "identity":
if value is not None:
raise ValueError("identity transform value must be null")
return image.clone()
pil = TF.to_pil_image(image)
if kind == "jpeg":
quality = int(value)
if not 1 <= quality <= 100:
raise ValueError("JPEG quality is outside its valid range")
buffer = io.BytesIO()
pil.save(buffer, format="JPEG", quality=quality)
buffer.seek(0)
with Image.open(buffer) as decoded:
return TF.to_tensor(decoded.convert("RGB"))
if kind == "resize":
scale = float(value)
if not 0.0 < scale <= 1.0:
raise ValueError("resize scale is outside (0,1]")
width, height = pil.size
reduced = pil.resize(
(max(1, round(width * scale)), max(1, round(height * scale))),
Image.Resampling.LANCZOS,
)
return TF.to_tensor(reduced.resize((width, height), Image.Resampling.LANCZOS))
raise ValueError("unsupported transformation kind")
def run_gate(
input_path: Path,
output_path: Path,
policy_path: Path,
policy_bundle: Path,
policy_identity: str,
oidc_issuer: str,
) -> dict:
if input_path.suffix.lower() not in {".jpg", ".jpeg", ".png", ".webp"}:
return {"status": "NOT_APPLICABLE", "reason": "unsupported_media_container"}
policy, policy_sha256 = verified_policy(
policy_path, policy_bundle, policy_identity, oidc_issuer
)
with Image.open(input_path) as source:
image = TF.to_tensor(source.convert("RGB")).unsqueeze(0)
if image.shape[2] < 64 or image.shape[3] < 64:
raise InsufficientData("asset is below the policy-supported image size")
model = videoseal.load(policy["model_name"])
model.eval()
if loaded_state_sha256(model) != policy["loaded_state_sha256"]:
raise ValueError("actual loaded watermark state differs from policy")
scaling_w = float(policy["scaling_w"])
if not 0.0 < scaling_w <= 1.0:
raise ValueError("embedding strength is invalid")
model.blender.scaling_w = scaling_w
with torch.inference_mode():
embedded = model.embed(image)
watermarked = embedded["imgs_w"][0].detach().cpu().clamp(0, 1)
message = embedded["msgs"][0].detach().cpu().flatten()
if message.numel() == 0:
raise InsufficientData("embedder did not return a payload")
output_path.parent.mkdir(parents=True, exist_ok=True)
TF.to_pil_image(watermarked).save(output_path)
output_bytes = output_path.read_bytes()
rows = []
for spec in policy["required_transforms"]:
candidate = transform_image(watermarked, spec).unsqueeze(0)
with torch.inference_mode():
detected = model.detect(candidate)["preds"][0].detach().cpu()
recovered = (detected[1:] > 0).to(message.dtype)
if recovered.numel() != message.numel():
raise InsufficientData("detector payload width differs from embedder")
rows.append({
"transform_id": spec["id"],
"presence_logit": float(detected[0]),
"bit_accuracy": float((recovered == message).float().mean()),
})
if len({row["transform_id"] for row in rows}) != len(policy["required_transforms"]):
raise InsufficientData("transformation results are missing or duplicated")
failures = [
row["transform_id"] for row in rows
if row["presence_logit"] < float(policy["minimum_presence_logit"])
or row["bit_accuracy"] < float(policy["minimum_bit_accuracy"])
]
return {
"schema_version": "aidefend.media-watermark-evidence.v1",
"status": "FAIL" if failures else "PASS",
"policy_version": policy["policy_version"],
"policy_sha256": policy_sha256,
"input_sha256": sha256_bytes(input_path.read_bytes()),
"output_sha256": sha256_bytes(output_bytes),
"loaded_state_sha256": policy["loaded_state_sha256"],
"payload_sha256": sha256_bytes(message.numpy().tobytes()),
"failed_transforms": failures,
"results": rows,
}
def run_with_status(**kwargs) -> dict:
try:
return run_gate(**kwargs)
except InsufficientData as exc:
return {"status": "INSUFFICIENT_DATA", "reason": str(exc)}
except Exception as exc:
return {"status": "ERROR", "reason": type(exc).__name__}
Verification, evidence, and outcomes
Execute the gate for every asset in the signed release population, then have a separate runner re-hash the input/output bytes, reload the pinned model state, re-run every policy transformation, and sign the canonical evidence document into append-only storage. Retain the immutable worker-image digest, dependency lock, signed policy and bundle, model-state digest, asset IDs, payload digests, raw detector values, complete population manifest, and independent readback. Do not disclose payload bits or calibration material outside the watermark service.
Action: Embed with the exact policy-bound VideoSeal state in the export worker, validate the complete transformation population, and release only after an independent replay produces signed evidence for the exact asset bytes.
AID-H-013.001-G002
Apply the maintained Hugging Face SynthID Text generation path and a policy-bound detector whose calibration and artifacts are independently verified.
Use this control only for a supported autoregressive Transformers generation stack where the team owns both the generation configuration and a calibrated SynthID Text detector. A signature-verified policy binds the exact model, tokenizer, detector, Transformers version, watermark configuration, minimum scorable length, calibration population, false-positive ceiling, true-positive floor, and quality limits. Keys remain in the watermark service and are never copied into public evidence.
Runnable maintained SynthID Text path
# File: watermarking/synthid_text_release.py
from __future__ import annotations
import math
import hashlib
import json
import os
import subprocess
import tempfile
from importlib.metadata import version
from pathlib import Path
import torch
from transformers import (
AutoModelForCausalLM,
AutoTokenizer,
BayesianDetectorModel,
SynthIDTextWatermarkDetector,
SynthIDTextWatermarkingConfig,
SynthIDTextWatermarkLogitsProcessor,
)
def _required_positive_seconds(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number of seconds")
return value
SYNTHID_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_SYNTHID_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS")
class InsufficientData(RuntimeError):
pass
def sha256(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def verified_json(path: Path, signature: Path, public_key: Path) -> tuple[dict, str]:
raw = path.read_bytes()
sig = signature.read_bytes()
if not raw or not sig:
raise ValueError("signed JSON or signature is empty")
with tempfile.TemporaryDirectory(prefix="synthid-verify-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
snapshot = root / "artifact.json"
signature_snapshot = root / "artifact.sig"
snapshot.write_bytes(raw)
signature_snapshot.write_bytes(sig)
os.chmod(snapshot, 0o400)
os.chmod(signature_snapshot, 0o400)
subprocess.run([
"cosign", "verify-blob", "--key", str(public_key),
"--bundle", str(signature_snapshot), str(snapshot),
], check=True, capture_output=True, text=True, timeout=SYNTHID_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS)
if snapshot.read_bytes() != raw:
raise ValueError("verified artifact changed")
return json.loads(raw.decode("utf-8", errors="strict")), sha256(raw)
def tree_sha256(root: Path) -> str:
rows = []
for path in sorted(item for item in root.rglob("*") if item.is_file()):
rows.append({
"path": path.relative_to(root).as_posix(),
"sha256": sha256(path.read_bytes()),
})
if not rows:
raise InsufficientData("artifact tree is empty")
return sha256(json.dumps(rows, sort_keys=True, separators=(",", ":")).encode())
def load_contract(
policy_path: Path,
policy_signature: Path,
policy_key: Path,
receipt_path: Path,
receipt_signature: Path,
receipt_key: Path,
keys_path: Path,
) -> tuple[dict, list[int], str]:
policy, policy_sha256 = verified_json(policy_path, policy_signature, policy_key)
required = {
"schema_version", "policy_version", "transformers_version",
"model_dir", "model_tree_sha256", "detector_dir",
"detector_tree_sha256", "keys_sha256", "synthid_config",
"generation_config", "minimum_generated_tokens",
"calibration_population_sha256", "maximum_false_positive_rate",
"minimum_true_positive_rate", "maximum_quality_delta",
"minimum_detection_probability",
}
if not isinstance(policy, dict) or set(policy) != required:
raise ValueError("text-watermark policy schema differs")
if policy["schema_version"] != "aidefend.synthid-text-policy.v1":
raise ValueError("unsupported text-watermark policy")
if version("transformers") != policy["transformers_version"]:
raise ValueError("Transformers version differs")
threshold = policy["minimum_detection_probability"]
if (
isinstance(threshold, bool)
or not math.isfinite(float(threshold))
or not 0.0 < float(threshold) < 1.0
):
raise ValueError("minimum detection probability is invalid")
keys_raw = keys_path.read_bytes()
if sha256(keys_raw) != policy["keys_sha256"]:
raise ValueError("watermark key digest differs")
keys = json.loads(keys_raw.decode("utf-8", errors="strict"))
if not isinstance(keys, list) or not keys or any(
isinstance(value, bool) or not isinstance(value, int) for value in keys
):
raise ValueError("watermark keys must be a non-empty integer list")
receipt, receipt_sha256 = verified_json(
receipt_path, receipt_signature, receipt_key
)
receipt_fields = {
"schema_version", "status", "policy_sha256",
"calibration_population_sha256", "model_tree_sha256",
"detector_tree_sha256", "false_positive_rate",
"true_positive_rate", "quality_delta",
}
if not isinstance(receipt, dict) or set(receipt) != receipt_fields:
raise InsufficientData("calibration receipt schema differs")
if (
receipt["schema_version"] != "aidefend.synthid-calibration.v1"
or receipt["status"] != "PASS"
or receipt["policy_sha256"] != policy_sha256
or receipt["calibration_population_sha256"]
!= policy["calibration_population_sha256"]
):
raise InsufficientData("calibration receipt is incomplete or unbound")
if (
float(receipt["false_positive_rate"])
> float(policy["maximum_false_positive_rate"])
or float(receipt["true_positive_rate"])
< float(policy["minimum_true_positive_rate"])
or abs(float(receipt["quality_delta"]))
> float(policy["maximum_quality_delta"])
):
raise ValueError("calibration thresholds are not satisfied")
return policy, keys, receipt_sha256
def generate_and_detect(
prompt: str,
policy_path: Path,
policy_signature: Path,
policy_key: Path,
receipt_path: Path,
receipt_signature: Path,
receipt_key: Path,
keys_path: Path,
) -> dict:
policy, keys, receipt_sha256 = load_contract(
policy_path, policy_signature, policy_key,
receipt_path, receipt_signature, receipt_key, keys_path,
)
model_dir = Path(policy["model_dir"]).resolve(strict=True)
detector_dir = Path(policy["detector_dir"]).resolve(strict=True)
if tree_sha256(model_dir) != policy["model_tree_sha256"]:
raise ValueError("model artifact tree differs")
if tree_sha256(detector_dir) != policy["detector_tree_sha256"]:
raise ValueError("detector artifact tree differs")
tokenizer = AutoTokenizer.from_pretrained(
model_dir, local_files_only=True, trust_remote_code=False
)
model = AutoModelForCausalLM.from_pretrained(
model_dir, local_files_only=True, trust_remote_code=False,
use_safetensors=True,
)
detector_model = BayesianDetectorModel.from_pretrained(
detector_dir, local_files_only=True
)
model.eval()
detector_model.eval()
device = next(model.parameters()).device
synthid_values = dict(policy["synthid_config"])
watermark_config = SynthIDTextWatermarkingConfig(
keys=keys, **synthid_values
)
detector_values = dict(detector_model.config.watermarking_config)
expected_values = {"keys": keys, **synthid_values}
if detector_values != expected_values:
raise ValueError("detector watermark configuration differs")
processor = SynthIDTextWatermarkLogitsProcessor(
**detector_values, device=str(device)
)
detector = SynthIDTextWatermarkDetector(
detector_model, processor, tokenizer
)
generation = dict(policy["generation_config"])
if set(generation) != {"max_new_tokens", "temperature", "top_p"}:
raise ValueError("generation configuration schema differs")
if int(generation["max_new_tokens"]) <= 0:
raise ValueError("max_new_tokens must be positive")
inputs = tokenizer(prompt, return_tensors="pt").to(device)
prompt_length = inputs["input_ids"].shape[1]
with torch.inference_mode():
output = model.generate(
**inputs,
watermarking_config=watermark_config,
do_sample=True,
max_new_tokens=int(generation["max_new_tokens"]),
temperature=float(generation["temperature"]),
top_p=float(generation["top_p"]),
)
generated_ids = output[:, prompt_length:]
if generated_ids.shape[1] < int(policy["minimum_generated_tokens"]):
raise InsufficientData("generated text is too short for the detector")
with torch.inference_mode():
prediction = detector(generated_ids)
posterior_probability = float(prediction[0].detach().cpu().item())
if not math.isfinite(posterior_probability) or not 0.0 <= posterior_probability <= 1.0:
raise ValueError("SynthID posterior probability is invalid")
detected = posterior_probability >= float(policy["minimum_detection_probability"])
token_bytes = generated_ids.detach().cpu().contiguous().numpy().tobytes()
return {
"schema_version": "aidefend.synthid-text-evidence.v1",
"status": "PASS" if detected else "FAIL",
"policy_version": policy["policy_version"],
"model_tree_sha256": policy["model_tree_sha256"],
"detector_tree_sha256": policy["detector_tree_sha256"],
"calibration_receipt_sha256": receipt_sha256,
"prompt_sha256": sha256(prompt.encode()),
"generated_token_ids_sha256": sha256(token_bytes),
"generated_token_count": int(generated_ids.shape[1]),
"detected": detected,
"posterior_probability": posterior_probability,
"detection_threshold": float(policy["minimum_detection_probability"]),
}
def run_with_status(**kwargs) -> dict:
try:
return generate_and_detect(**kwargs)
except InsufficientData as exc:
return {"status": "INSUFFICIENT_DATA", "reason": str(exc)}
except Exception as exc:
return {"status": "ERROR", "reason": type(exc).__name__}
Calibration dependency, verification, and evidence
The independent calibration runner is the required upstream producer. Before deployment it signs aidefend.synthid-calibration.v1 over the exact policy digest, model and detector tree digests, complete positive/negative/quality population digest, per-language and per-length results, false-positive rate, true-positive rate, and quality delta. The serving boundary re-verifies that signature and every binding; an upstream PASS is only an input and never becomes this guidance's PASS. Retain the signed policy and calibration receipt, verification records, dependency lock, immutable service-image digest, exact model/tokenizer/detector trees, key digest without key material, request/output token digests, token count, and detector result in append-only storage. A separate runner reloads the same artifacts and replays policy-selected generated samples.
Action: Use the maintained Transformers SynthID Text APIs with exact offline artifacts, admit only a signature-verified independent calibration receipt, and promote the service only after a separate replay verifies the complete eligible population.
AID-H-013.001-G003
Attach C2PA Content Credentials with a managed signer and admit them only after pinned trust-list verification and independent tamper replay.
Use this control for release containers supported by the pinned C2PA Tool version and for which the organization has an enterprise C2PA signing certificate plus a KMS/HSM-backed signer. A signature-verified release policy binds the exact source asset, release receipt, claim generator, digital source type, external signer executable digest and identity, C2PA Tool version, trust-anchor snapshot digest and freshness, supported-container population, and required negative tests. Never rely on the tool's built-in test certificate or on a non-empty manifest as a trust verdict.
Managed signing and trust-verifying release gate
# File: provenance/c2pa_release_gate.py
from __future__ import annotations
import math
import hashlib
import json
import os
import shlex
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
def _required_positive_seconds(name: str) -> float:
value = float(os.environ[name])
if not math.isfinite(value) or value <= 0:
raise RuntimeError(f"{name} must be a finite positive number of seconds")
return value
C2PA_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_C2PA_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS")
C2PA_TRUST_REPORT_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_C2PA_TRUST_REPORT_TIMEOUT_SECONDS")
C2PA_VERSION_PROBE_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_C2PA_VERSION_PROBE_TIMEOUT_SECONDS")
C2PA_MANIFEST_CREATE_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H013_C2PA_MANIFEST_CREATE_TIMEOUT_SECONDS")
class InsufficientData(RuntimeError):
pass
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def verified_json(
path: Path, signature: Path, public_key: Path
) -> tuple[dict, str]:
raw = path.read_bytes()
sig = signature.read_bytes()
if not raw or not sig:
raise ValueError("signed JSON or signature is empty")
with tempfile.TemporaryDirectory(prefix="c2pa-verify-") as directory:
root = Path(directory)
os.chmod(root, 0o700)
snapshot = root / "artifact.json"
signature_snapshot = root / "artifact.sig"
snapshot.write_bytes(raw)
signature_snapshot.write_bytes(sig)
os.chmod(snapshot, 0o400)
os.chmod(signature_snapshot, 0o400)
subprocess.run([
"cosign", "verify-blob", "--key", str(public_key),
"--bundle", str(signature_snapshot), str(snapshot),
], check=True, capture_output=True, text=True, timeout=C2PA_CONTROL_JSON_VERIFY_TIMEOUT_SECONDS)
if snapshot.read_bytes() != raw:
raise ValueError("verified JSON changed")
return (
json.loads(raw.decode("utf-8", errors="strict")),
hashlib.sha256(raw).hexdigest(),
)
def load_contract(
policy_path: Path,
policy_signature: Path,
policy_key: Path,
receipt_path: Path,
receipt_signature: Path,
receipt_key: Path,
source_path: Path,
trust_anchors: Path,
) -> tuple[dict, dict, str, str]:
policy, policy_sha256 = verified_json(
policy_path, policy_signature, policy_key
)
required = {
"schema_version", "policy_version", "source_sha256",
"release_receipt_sha256", "receipt_not_before",
"receipt_not_after", "claim_generator_name",
"claim_generator_version", "digital_source_type",
"c2patool_version", "signer_executable",
"signer_executable_sha256", "signer_arguments",
"trust_anchors_sha256", "trust_anchors_not_after",
"supported_extensions", "required_negative_fixtures",
}
if not isinstance(policy, dict) or set(policy) != required:
raise ValueError("C2PA policy schema differs")
if policy["schema_version"] != "aidefend.c2pa-release-policy.v1":
raise ValueError("unsupported C2PA policy")
if source_path.suffix.lower() not in set(policy["supported_extensions"]):
return policy, {}, policy_sha256, "NOT_APPLICABLE"
if sha256_file(source_path) != policy["source_sha256"]:
raise ValueError("source asset digest differs")
if sha256_file(trust_anchors) != policy["trust_anchors_sha256"]:
raise ValueError("trust-anchor snapshot digest differs")
now = datetime.now(timezone.utc)
trust_expiry = datetime.fromisoformat(
policy["trust_anchors_not_after"].replace("Z", "+00:00")
)
if now > trust_expiry:
raise InsufficientData("trust-anchor snapshot is stale")
receipt, receipt_sha256 = verified_json(
receipt_path, receipt_signature, receipt_key
)
if receipt_sha256 != policy["release_receipt_sha256"]:
raise ValueError("release receipt digest differs")
receipt_required = {
"schema_version", "release_id", "source_sha256",
"generator_model_sha256", "generation_policy_sha256",
"produced_at",
}
if not isinstance(receipt, dict) or set(receipt) != receipt_required:
raise InsufficientData("release receipt schema differs")
if (
receipt["schema_version"] != "aidefend.generated-asset-receipt.v1"
or receipt["source_sha256"] != policy["source_sha256"]
):
raise ValueError("release receipt binding differs")
not_before = datetime.fromisoformat(
policy["receipt_not_before"].replace("Z", "+00:00")
)
not_after = datetime.fromisoformat(
policy["receipt_not_after"].replace("Z", "+00:00")
)
produced_at = datetime.fromisoformat(
receipt["produced_at"].replace("Z", "+00:00")
)
if not not_before <= produced_at <= not_after:
raise ValueError("release receipt is outside the policy window")
return policy, receipt, policy_sha256, receipt_sha256
def collect_status_codes(value: object) -> list[str]:
codes = []
if isinstance(value, dict):
for key, child in value.items():
if key == "validation_status" and isinstance(child, list):
for row in child:
if isinstance(row, dict):
codes.append(str(row.get("code", "unknown_validation_error")))
else:
codes.append(str(row))
else:
codes.extend(collect_status_codes(child))
elif isinstance(value, list):
for child in value:
codes.extend(collect_status_codes(child))
return codes
def trust_report(asset: Path, trust_anchors: Path) -> tuple[dict, list[str], int]:
process = subprocess.run([
"c2patool", str(asset), "trust",
"--trust_anchors", str(trust_anchors),
], check=False, capture_output=True, text=True, timeout=C2PA_TRUST_REPORT_TIMEOUT_SECONDS)
try:
report = json.loads(process.stdout)
except json.JSONDecodeError as exc:
raise ValueError("c2patool trust output is not JSON") from exc
codes = collect_status_codes(report)
if not report.get("active_manifest"):
codes.append("missing_active_manifest")
if process.returncode != 0:
codes.append("c2patool_nonzero")
return report, sorted(set(codes)), process.returncode
def run_gate(
source_path: Path,
output_path: Path,
policy_path: Path,
policy_signature: Path,
policy_key: Path,
receipt_path: Path,
receipt_signature: Path,
receipt_key: Path,
trust_anchors: Path,
) -> dict:
policy, receipt, policy_sha256, receipt_binding = load_contract(
policy_path, policy_signature, policy_key,
receipt_path, receipt_signature, receipt_key,
source_path, trust_anchors,
)
if receipt_binding == "NOT_APPLICABLE":
return {"status": "NOT_APPLICABLE", "reason": "unsupported_container"}
tool_version = subprocess.run(
["c2patool", "-V"], check=True, capture_output=True,
text=True, timeout=C2PA_VERSION_PROBE_TIMEOUT_SECONDS,
).stdout.strip()
if tool_version != policy["c2patool_version"]:
raise ValueError("c2patool version differs")
signer = Path(policy["signer_executable"]).resolve(strict=True)
if sha256_file(signer) != policy["signer_executable_sha256"]:
raise ValueError("external signer executable digest differs")
signer_arguments = policy["signer_arguments"]
if not isinstance(signer_arguments, list) or any(
not isinstance(value, str) for value in signer_arguments
):
raise ValueError("signer arguments must be strings")
signer_command = shlex.join([str(signer), *signer_arguments])
source_type = policy["digital_source_type"]
allowed_source_types = {
"http://cv.iptc.org/newscodes/digitalsourcetype/algorithmicMedia",
"http://cv.iptc.org/newscodes/digitalsourcetype/trainedAlgorithmicMedia",
"http://cv.iptc.org/newscodes/digitalsourcetype/compositeWithTrainedAlgorithmicMedia",
}
if source_type not in allowed_source_types:
raise ValueError("digital source type must be a canonical IPTC URI")
source_type_cli = source_type.rsplit("/", 1)[-1]
manifest = {
"claim_generator": (
policy["claim_generator_name"] + " "
+ policy["claim_generator_version"]
),
"claim_generator_info": [{
"name": policy["claim_generator_name"],
"version": policy["claim_generator_version"],
}],
"assertions": [],
}
if output_path.exists():
raise ValueError("refusing to overwrite an existing signed asset")
output_path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="c2pa-sign-") as directory:
manifest_path = Path(directory) / "manifest.json"
manifest_path.write_text(
json.dumps(manifest, sort_keys=True, separators=(",", ":")),
encoding="utf-8",
)
subprocess.run([
"c2patool", str(source_path),
"--manifest", str(manifest_path),
"--output", str(output_path),
"--signer-path", signer_command,
"--create", source_type_cli,
], check=True, capture_output=True, text=True, timeout=C2PA_MANIFEST_CREATE_TIMEOUT_SECONDS)
report, validation_codes, _ = trust_report(output_path, trust_anchors)
active = report.get("active_manifest")
manifests = report.get("manifests")
if (
not isinstance(active, str)
or not isinstance(manifests, dict)
or active not in manifests
or not isinstance(manifests[active].get("signature_info"), dict)
):
validation_codes.append("missing_signature_info")
fixture_rows = policy["required_negative_fixtures"]
if not isinstance(fixture_rows, list) or not fixture_rows:
raise InsufficientData("negative fixture population is empty")
fixture_ids = [row.get("id") for row in fixture_rows if isinstance(row, dict)]
if len(fixture_ids) != len(fixture_rows) or len(set(fixture_ids)) != len(fixture_ids):
raise InsufficientData("negative fixture IDs are missing or duplicated")
negative_failures = []
negative_evidence = []
for fixture in fixture_rows:
if set(fixture) != {"id", "path", "sha256", "expected_status_code"}:
raise InsufficientData("negative fixture schema differs")
fixture_path = Path(fixture["path"]).resolve(strict=True)
if sha256_file(fixture_path) != fixture["sha256"]:
raise ValueError("negative fixture digest differs")
fixture_report, fixture_codes, returncode = trust_report(
fixture_path, trust_anchors
)
expected = fixture["expected_status_code"]
if expected not in fixture_codes:
negative_failures.append(fixture["id"])
negative_evidence.append({
"id": fixture["id"], "sha256": fixture["sha256"],
"codes": fixture_codes, "returncode": returncode,
"report_sha256": hashlib.sha256(
json.dumps(
fixture_report, sort_keys=True, separators=(",", ":")
).encode()
).hexdigest(),
})
failures = sorted(set(validation_codes))
return {
"schema_version": "aidefend.c2pa-release-evidence.v1",
"status": "FAIL" if failures or negative_failures else "PASS",
"policy_version": policy["policy_version"],
"policy_sha256": policy_sha256,
"release_receipt_sha256": receipt_binding,
"release_id": receipt["release_id"],
"source_sha256": policy["source_sha256"],
"signed_asset_sha256": sha256_file(output_path),
"c2patool_version": tool_version,
"signer_executable_sha256": policy["signer_executable_sha256"],
"trust_anchors_sha256": policy["trust_anchors_sha256"],
"active_manifest": active,
"validation_report_sha256": hashlib.sha256(
json.dumps(
report, sort_keys=True, separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
).hexdigest(),
"validation_codes": failures,
"negative_fixture_failures": negative_failures,
"negative_evidence": negative_evidence,
}
def run_with_status(**kwargs) -> dict:
try:
return run_gate(**kwargs)
except InsufficientData as exc:
return {"status": "INSUFFICIENT_DATA", "reason": str(exc)}
except Exception as exc:
return {"status": "ERROR", "reason": type(exc).__name__}
Independent verification, evidence, and outcomes
The release receipt is a required upstream dependency produced by the generation pipeline and signed under a different identity; the C2PA gate verifies its schema, digest, asset binding, model/policy digests, and policy-owned time window before signing. The downstream PASS remains distinct. Preserve the signed source, signed release receipt, policy and verification records, exact external-signer executable digest and signer identity, C2PA Tool version, manifest definition, signed-asset digest, pinned trust-anchor bytes and freshness, complete validation JSON digest, every tampered/stripped/untrusted negative fixture and result, and independent replay receipt in append-only storage. A separate verifier re-runs trust validation and the complete negative population using an identity that cannot invoke the signer.
Action: Sign each eligible release asset through the digest-pinned external KMS/HSM signer, validate it against the pinned current trust list, and release only after an independent runner reproduces trust success plus every required negative-test failure.
Open-source tools
- Hugging Face Transformers
- c2patool
- c2pa-rs
- Cosign
- PyTorch
- torchvision
- Pillow
- TrustMark
Source-available / open-weight tools
- VideoSeal (MIT; open-weight)
- PixelSeal (MIT; open-weight)
Commercial / hosted tools
- Adobe Firefly Content Credentials
- Google Vertex AI Imagen SynthID
Threat mappings
MITRE ATLAS
- N/A
MAESTRO
- N/A
OWASP LLM Top 10 2026
- LLM07:2026 Misinformation (signed provenance lets consumers identify generated or altered media and reduces false attribution, but it does not prove factual truth)
OWASP ML Top 10 2023
- ML09:2023 Output Integrity Attack
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
- Model Serving - Inference response 10.2: Output manipulation