AID-D-001.002: Synthetic Media & Deepfake Forensics

Version 1.20261015 · Actionable sub-technique

Detects manipulated or synthetically generated media (e.g., deepfakes) by performing a forensic analysis that identifies a combination of specific technical artifacts and inconsistencies.

Signals fused: image compression anomalies, unnatural biological signals (blinking, vocal patterns), audio-visual mismatches, and reconstruction-error anomaly signals from trusted inversion baselines.

Use as triage: treat the result as a forensic signal, not a standalone identity or authorization decision; high-impact workflows still need out-of-band verification, provenance checks, and dual-control procedures.

Pillars: data, app

Lifecycle: validation, operation

Scope boundary

Owns forensic synthetic-media and manipulation analysis within image, audio, video, biometric, temporal, and related authenticity signals. Its result is triage evidence, not standalone identity or authorization proof.

AID-D-007.001: AID-D-001.002 determines whether a media signal appears synthetic or manipulated; AID-D-007.001 detects cross-modal contradiction, OCR or QR instructions, hidden payloads, and artifact-level inconsistency in a bounded batch. Authenticity of one signal and consistency across media or extracted payloads are separate forensic questions.

AID-M-003.005: AID-D-001.002 applies reconstruction and other forensic signals to runtime media and emits findings; AID-M-003.005 creates and recalibrates the trusted reconstruction-error reference. The detector consumes the approved baseline and must not silently redefine it from suspicious ingress traffic.

Warning

Unreliable as Sole Authentication or Authorization Control

Synthetic-media detectors are useful for triage, prioritization, and evidence gathering, but current deepfake detection remains brittle under new generators, compression, recapture, adversarial post-processing, and low-quality media. Do not allow or deny money movement, account recovery, credential reset, executive instruction, hiring, law-enforcement, or safety-critical action based only on a detector score. Pair this technique with process controls such as independent callback verification, dual authorization, signed provenance / C2PA-style source evidence where available, and HITL escalation for ambiguous cases.

Implementation guidance

AID-D-001.002-G001

Analyze for digital manipulation artifacts in images.
Reusable forensic module

Error-level analysis is JPEG-specific and must remain a heuristic signal; it does not prove that an image is authentic or manipulated. Process immutable bytes in memory, reject decompression bombs, derive policy-controlled numeric features, and fuse the result with the other synthetic-media signals in this sub-technique. Do not write attacker-controlled media to a shared temporary filename or require manual image inspection.

from __future__ import annotations

# File: detection/jpeg_ela.py

import hashlib
from dataclasses import asdict, dataclass
from io import BytesIO
from typing import Literal

import numpy as np
from PIL import Image, ImageChops, ImageOps, UnidentifiedImageError

ControlStatus = Literal["PASS", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"]
FindingStatus = Literal["FINDING", "NO_FINDING", "UNKNOWN"]


@dataclass(frozen=True)
class ElaPolicy:
    policy_version: str
    jpeg_quality: int
    patch_size: int
    stride: int
    max_pixels: int
    max_patch_variance: float
    max_difference_p99: float


def evaluate_jpeg_ela(artifact_id: str, image_bytes: bytes, policy: ElaPolicy) -> dict:
    digest = hashlib.sha256(image_bytes).hexdigest()
    if not (1 <= policy.jpeg_quality <= 95):
        raise ValueError("jpeg_quality outside the signed policy range")
    if min(policy.patch_size, policy.stride, policy.max_pixels) < 1:
        raise ValueError("invalid signed ELA bounds")
    try:
        source = Image.open(BytesIO(image_bytes))
        if source.format != "JPEG":
            return {
                "control_status": "NOT_APPLICABLE",
                "finding_status": "UNKNOWN",
                "reason": "artifact is not JPEG",
                "artifact_id": artifact_id,
                "artifact_sha256": digest,
                "policy_version": policy.policy_version,
            }
        width, height = source.size
        if width * height > policy.max_pixels:
            return {
                "control_status": "INSUFFICIENT_DATA",
                "finding_status": "UNKNOWN",
                "reason": "image exceeds signed decode bound",
                "artifact_id": artifact_id,
                "artifact_sha256": digest,
                "policy_version": policy.policy_version,
            }
        original = ImageOps.exif_transpose(source).convert("RGB")
        encoded = BytesIO()
        original.save(encoded, format="JPEG", quality=policy.jpeg_quality)
        encoded.seek(0)
        resaved = Image.open(encoded).convert("RGB")
        difference = np.asarray(
            ImageChops.difference(original, resaved), dtype=np.float32
        )
    except (UnidentifiedImageError, OSError, ValueError) as exc:
        return {
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "reason": type(exc).__name__,
            "artifact_id": artifact_id,
            "artifact_sha256": digest,
            "policy_version": policy.policy_version,
        }

    grayscale = difference.mean(axis=2)
    variances: list[float] = []
    for y in range(0, height, policy.stride):
        for x in range(0, width, policy.stride):
            patch = grayscale[y:min(y + policy.patch_size, height), x:min(x + policy.patch_size, width)]
            if patch.shape == (policy.patch_size, policy.patch_size):
                variances.append(float(np.var(patch)))
    if not variances:
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "reason": "no complete analysis patch",
            "artifact_id": artifact_id,
            "artifact_sha256": digest,
            "policy_version": policy.policy_version,
        }
    features = {
        "max_patch_variance": round(max(variances), 6),
        "difference_p99": round(float(np.percentile(grayscale, 99)), 6),
        "patch_count": len(variances),
    }
    reasons = []
    if features["max_patch_variance"] >= policy.max_patch_variance:
        reasons.append("patch_variance_threshold")
    if features["difference_p99"] >= policy.max_difference_p99:
        reasons.append("difference_p99_threshold")
    return {
        "schema_version": "aidefend.jpeg_ela_finding.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if reasons else "NO_FINDING",
        "reasons": reasons,
        "artifact_id": artifact_id,
        "artifact_sha256": digest,
        "policy_version": policy.policy_version,
        "features": features,
    }

Verification: Calibrate thresholds per camera, codec, and ingest pipeline on a signed corpus. Replay original-camera JPEGs, edited JPEGs, recompressed social-media images, non-JPEG files, corrupt files, oversized files, and very small images; verify deterministic features and explicit non-clean states.

Action: Run this bounded module for every applicable JPEG artifact, retain only digests and numeric features, and forward its finding as one forensic signal rather than an authenticity verdict.

AID-D-001.002-G002

Analyze for anomalous visual liveness signals in video.
Concept:

Deepfake video often fails to preserve natural blink cadence, eye closure duration, and other face-liveness patterns over time. Run liveness checks only after you have a stable face track; otherwise motion blur, occlusion, or profile views will create noise.

Step 1: Track eye landmarks and summarize blink behavior
from __future__ import annotations

# File: detection/video_liveness.py
import hashlib
import json
import math
import os
import subprocess
from dataclasses import asdict, dataclass
from pathlib import Path
import hmac
import re
import tempfile




def _aidefend_write_private(path: Path, payload: bytes) -> None:
    descriptor = os.open(
        path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600,
    )
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _aidefend_strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str,
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise RuntimeError("signed payload/proof inputs are invalid")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "proof"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(proof_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw


from scipy.spatial import distance as dist


def load_signed_video_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["D001002_VIDEO_POLICY_PATH"])
    bundle_path = Path(os.environ["D001002_VIDEO_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = _aidefend_capture_signed_bytes(
        policy_path, bundle_path, os.environ["D001002_VIDEO_POLICY_VERIFY_KEY"],
        verify_timeout, "--bundle",
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "closed_ear_threshold",
        "minimum_blinks_per_minute", "maximum_closed_run_frames",
        "minimum_observation_seconds",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed video-liveness policy fields differ")
    if (
        policy["schema_version"] != "aidefend.video-liveness-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signed video-liveness policy identity is invalid")
    for field in (
        "closed_ear_threshold", "minimum_blinks_per_minute",
        "minimum_observation_seconds",
    ):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or float(value) <= 0
        ):
            raise RuntimeError("signed video-liveness bound is invalid: " + field)
    closed_frames = policy["maximum_closed_run_frames"]
    if isinstance(closed_frames, bool) or not isinstance(closed_frames, int) or closed_frames < 1:
        raise RuntimeError("signed closed-frame bound is invalid")
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


VIDEO_POLICY, VIDEO_POLICY_SHA256 = load_signed_video_policy()


@dataclass(frozen=True)
class BlinkMetrics:
    blink_count: int
    longest_closed_run_frames: int
    mean_blink_duration_frames: float


def calculate_ear(eye_points) -> float:
    a = dist.euclidean(eye_points[1], eye_points[5])
    b = dist.euclidean(eye_points[2], eye_points[4])
    c = dist.euclidean(eye_points[0], eye_points[3])
    if not math.isfinite(c) or c <= 0:
        raise ValueError("eye landmarks do not define a valid horizontal span")
    value = (a + b) / (2.0 * c)
    if not math.isfinite(value) or value < 0:
        raise ValueError("computed eye-aspect ratio is invalid")
    return value


def summarize_blinks(ear_series: list[float]) -> BlinkMetrics:
    if not ear_series or any(
        isinstance(value, bool)
        or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        or float(value) < 0
        for value in ear_series
    ):
        raise ValueError("EAR population is empty or malformed")
    threshold = float(VIDEO_POLICY["closed_ear_threshold"])
    blink_count = 0
    closed_run = 0
    longest_closed_run = 0
    durations: list[int] = []
    for raw_value in ear_series:
        value = float(raw_value)
        if value < threshold:
            closed_run += 1
            longest_closed_run = max(longest_closed_run, closed_run)
            continue
        if closed_run:
            blink_count += 1
            durations.append(closed_run)
            closed_run = 0
    if closed_run:
        durations.append(closed_run)
        longest_closed_run = max(longest_closed_run, closed_run)
    mean_duration = sum(durations) / len(durations) if durations else 0.0
    return BlinkMetrics(blink_count, longest_closed_run, mean_duration)


def evaluate_video_liveness(
    ear_series: list[float],
    duration_seconds: float,
) -> dict[str, object]:
    if (
        isinstance(duration_seconds, bool)
        or not isinstance(duration_seconds, (int, float))
        or not math.isfinite(float(duration_seconds))
        or float(duration_seconds) < float(VIDEO_POLICY["minimum_observation_seconds"])
    ):
        raise ValueError("video duration does not satisfy signed observation policy")
    metrics = summarize_blinks(ear_series)
    blinks_per_minute = metrics.blink_count * 60.0 / float(duration_seconds)
    anomalous = (
        blinks_per_minute < float(VIDEO_POLICY["minimum_blinks_per_minute"])
        or metrics.longest_closed_run_frames
        > VIDEO_POLICY["maximum_closed_run_frames"]
    )
    return {
        "schema_version": "aidefend.video-liveness-result.v1",
        "control": "AID-D-001.002",
        "policy_version": VIDEO_POLICY["policy_version"],
        "policy_sha256": VIDEO_POLICY_SHA256,
        "sample_count": len(ear_series),
        "duration_seconds": float(duration_seconds),
        "metrics": asdict(metrics),
        "blinks_per_minute": blinks_per_minute,
        "finding_status": "FINDING" if anomalous else "NO_FINDING",
    }
Step 2: Apply quality gates before alerting

Only score speaking or frontal-face segments with sufficient frame quality. Persist the blink metrics, face-track identifier, and frame window so analysts can reproduce the decision.

Action: Run visual liveness on face-tracked video segments, not on single frames. Flag videos with sustained blink suppression or implausible closure timing, and route low-quality segments to manual review instead of auto-failing them.

Verify safely

An independent verifier must re-run track extraction and liveness scoring on signed positive, benign, no-face, truncated, and unsupported-codec fixtures and compare track and window digests, read the stored objects back, verify signatures and digests, and compare the canonical result digest.

AID-D-001.002-G003

Analyze speech biometrics and acoustic artifacts to detect synthetic or cloned audio.
Concept:

Synthetic or cloned speech often exhibits abnormal pitch stability, spectral flatness, or transient artifacts that differ from live human speech. Evaluate short, speech-active windows and aggregate scores across the clip so one noisy segment does not dominate the verdict.

Step 1: Extract stable acoustic features
from __future__ import annotations

# File: detection/audio_liveness.py
import hashlib
import json
import math
import os
import subprocess
from pathlib import Path
import hmac
import re
import tempfile




def _aidefend_write_private(path: Path, payload: bytes) -> None:
    descriptor = os.open(
        path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600,
    )
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _aidefend_strict_json(raw: bytes) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str,
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise RuntimeError("signed payload/proof inputs are invalid")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "proof"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(proof_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw


import librosa
import numpy as np


def load_signed_audio_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["D001002_AUDIO_POLICY_PATH"])
    bundle_path = Path(os.environ["D001002_AUDIO_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = _aidefend_capture_signed_bytes(
        policy_path, bundle_path, os.environ["D001002_AUDIO_POLICY_VERIFY_KEY"],
        verify_timeout, "--bundle",
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "sample_rate_hz", "mfcc_count",
        "fft_window_size", "hop_length", "mel_band_count", "pitch_floor_hz",
        "pitch_ceiling_hz", "synthetic_probability_threshold",
        "synthetic_class_label",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed audio-liveness policy fields differ")
    if (
        policy["schema_version"] != "aidefend.audio-liveness-detection-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
        or not isinstance(policy["synthetic_class_label"], str)
        or not policy["synthetic_class_label"]
    ):
        raise RuntimeError("signed audio-liveness policy identity is invalid")
    for field in (
        "sample_rate_hz", "mfcc_count", "fft_window_size",
        "hop_length", "mel_band_count",
    ):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError("signed audio feature dimension is invalid: " + field)
    if policy["hop_length"] > policy["fft_window_size"]:
        raise RuntimeError("signed audio hop length exceeds FFT window")
    for field in ("pitch_floor_hz", "pitch_ceiling_hz"):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or float(value) <= 0
        ):
            raise RuntimeError("signed pitch bound is invalid: " + field)
    if not (
        float(policy["pitch_floor_hz"]) < float(policy["pitch_ceiling_hz"])
        < float(policy["sample_rate_hz"]) / 2
    ):
        raise RuntimeError("signed pitch range violates ordering or Nyquist")
    threshold = policy["synthetic_probability_threshold"]
    if (
        isinstance(threshold, bool)
        or not isinstance(threshold, (int, float))
        or not math.isfinite(float(threshold))
        or not 0 <= float(threshold) <= 1
    ):
        raise RuntimeError("signed audio-liveness threshold is invalid")
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


AUDIO_POLICY, AUDIO_POLICY_SHA256 = load_signed_audio_policy()


def extract_audio_features(audio_path: str) -> np.ndarray:
    signal, sample_rate = librosa.load(
        audio_path,
        sr=AUDIO_POLICY["sample_rate_hz"],
        mono=True,
    )
    if signal.size == 0 or not np.isfinite(signal).all():
        raise ValueError("decoded audio population is empty or non-finite")
    common = {
        "n_fft": AUDIO_POLICY["fft_window_size"],
        "hop_length": AUDIO_POLICY["hop_length"],
    }
    mfcc = librosa.feature.mfcc(
        y=signal,
        sr=sample_rate,
        n_mfcc=AUDIO_POLICY["mfcc_count"],
        n_mels=AUDIO_POLICY["mel_band_count"],
        **common,
    ).mean(axis=1)
    flatness = librosa.feature.spectral_flatness(y=signal, **common).mean(axis=1)
    zcr = librosa.feature.zero_crossing_rate(
        signal,
        frame_length=AUDIO_POLICY["fft_window_size"],
        hop_length=AUDIO_POLICY["hop_length"],
    ).mean(axis=1)
    pitch, _, _ = librosa.pyin(
        signal,
        fmin=float(AUDIO_POLICY["pitch_floor_hz"]),
        fmax=float(AUDIO_POLICY["pitch_ceiling_hz"]),
        sr=sample_rate,
        frame_length=AUDIO_POLICY["fft_window_size"],
        hop_length=AUDIO_POLICY["hop_length"],
    )
    pitch_std = np.nan_to_num(pitch, nan=0.0).std()
    vector = np.concatenate([mfcc, flatness, zcr, np.array([pitch_std])])
    if vector.ndim != 1 or not np.isfinite(vector).all():
        raise ValueError("audio-liveness feature vector is malformed")
    return vector


class AudioLivenessDetector:
    def __init__(self, classifier):
        classes = [str(value) for value in classifier.classes_]
        label = AUDIO_POLICY["synthetic_class_label"]
        if classes.count(label) != 1:
            raise RuntimeError("classifier lacks one policy-bound synthetic class")
        self.classifier = classifier
        self.synthetic_class_index = classes.index(label)

    def classify(self, feature_vector: np.ndarray) -> dict[str, object]:
        if feature_vector.ndim != 1 or not np.isfinite(feature_vector).all():
            raise ValueError("audio-liveness feature vector is malformed")
        probabilities = self.classifier.predict_proba([feature_vector])[0]
        synthetic_prob = float(probabilities[self.synthetic_class_index])
        if not math.isfinite(synthetic_prob) or not 0 <= synthetic_prob <= 1:
            raise RuntimeError("classifier returned an invalid probability")
        anomalous = synthetic_prob >= float(
            AUDIO_POLICY["synthetic_probability_threshold"]
        )
        return {
            "schema_version": "aidefend.audio-liveness-result.v1",
            "control": "AID-D-001.002",
            "policy_version": AUDIO_POLICY["policy_version"],
            "policy_sha256": AUDIO_POLICY_SHA256,
            "synthetic_probability": synthetic_prob,
            "finding_status": "FINDING" if anomalous else "NO_FINDING",
        }
Step 2: Aggregate per-window scores

Score multiple speech-active windows, discard low-SNR regions, and alert only when anomalous windows are sustained or clustered. Keep the raw feature summary and model version as evidence for later review.

Action: Use an audio-liveness classifier over normalized speech windows, aggregate the results across the clip, and flag segments whose synthetic probability remains above threshold after low-quality windows are excluded.

Verify safely

An independent verifier must re-decode and re-segment signed cloned, natural, noisy, silent, and unsupported fixtures and compare segment boundaries, feature digests, and findings, read the stored objects back, verify signatures and digests, and compare the canonical result digest.

AID-D-001.002-G004

Detect audio-visual synchronization anomalies to identify lip-sync deepfakes.
Concept:

Lip-sync deepfakes often preserve a believable face and believable audio independently, but the two streams drift when you compare mouth motion against phoneme or energy timing. This step is intentionally limited to synchronization analysis. Generic image-text or modality-pair semantic consistency belongs in AID-D-007.

Step 1: Compare mouth-motion and speech timing
from __future__ import annotations

# File: detection/lipsync_detector.py

import hmac as _aidefend_policy_hmac
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import re as _aidefend_policy_re
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_write_private(path, payload):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_WRONLY
        | _aidefend_policy_os.O_CREAT
        | _aidefend_policy_os.O_EXCL,
        0o600,
    )
    with _aidefend_policy_os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        _aidefend_policy_os.fsync(stream.fileno())
    _aidefend_policy_os.chmod(path, 0o400)


def _aidefend_strict_json(raw):
    def reject_duplicate_keys(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return _aidefend_policy_json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    if policy_path == signature_path:
        raise RuntimeError("policy and signature paths must differ")
    policy_raw = policy_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = _aidefend_policy_os.environ[key_env]
    remote_key = bool(_aidefend_policy_re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*:(?://)?", key_reference
    )) and not key_reference.startswith("-----BEGIN ")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else _AidefendPolicyPath(key_reference).read_bytes()
    )
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    with _aidefend_policy_tempfile.TemporaryDirectory(
        prefix="aidefend-signed-policy-"
    ) as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        policy_snapshot = root / "policy.json"
        signature_snapshot = root / "policy.sig"
        _aidefend_write_private(policy_snapshot, policy_raw)
        _aidefend_write_private(signature_snapshot, signature_raw)
        captured = [(policy_snapshot, policy_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            _aidefend_policy_subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(policy_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=verify_timeout,
            )
            if any(
                not _aidefend_policy_hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            policy = _aidefend_strict_json(policy_raw)
        finally:
            for path, _ in captured:
                _aidefend_policy_os.chmod(path, 0o600)
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy


import numpy as np


AUDIO_VIDEO_POLICY = _aidefend_load_signed_policy(
    path_env="D001002_AV_POLICY_PATH",
    signature_env="D001002_AV_POLICY_SIGNATURE",
    key_env="D001002_AV_POLICY_VERIFY_KEY",
    schema_version="aidefend.audio-video-correlation-policy.v1",
)
MINIMUM_AV_CORRELATION = AUDIO_VIDEO_POLICY.get("minimum_correlation")
if (
    isinstance(MINIMUM_AV_CORRELATION, bool)
    or not isinstance(MINIMUM_AV_CORRELATION, (int, float))
    or not -1 <= float(MINIMUM_AV_CORRELATION) <= 1
):
    raise RuntimeError("signed audio-video correlation threshold is invalid")


def normalized_correlation(a: np.ndarray, b: np.ndarray) -> float:
    a = (a - a.mean()) / (a.std() + 1e-8)
    b = (b - b.mean()) / (b.std() + 1e-8)
    return float(np.correlate(a, b, mode='valid')[0] / len(a))


def is_lipsync_anomalous(mouth_open_series: np.ndarray, audio_energy_series: np.ndarray) -> dict[str, float | bool]:
    if len(mouth_open_series) != len(audio_energy_series):
        raise ValueError('aligned mouth and audio windows are required')

    score = normalized_correlation(mouth_open_series, audio_energy_series)
    return {
        'correlation': score,
        'is_anomalous': score < float(MINIMUM_AV_CORRELATION),
    }
Step 2: Restrict scoring to speaking segments

Run synchronization checks only when a face track and speech activity are both present. Log the segment bounds, correlation score, and track identifier so investigators can replay the exact suspicious window.

Action: Use synchronized audio-energy and mouth-motion windows to flag lip-sync anomalies. Treat repeated low-correlation speaking segments as evidence of a likely audio-visual deepfake.

AID-D-001.002-G005

Score media against the currently valid signed reference using the same measured reconstruction pipeline and preprocessing, and emit anomaly findings above the approved threshold.
Consume the measured reconstruction pipeline

Use the exact shared reconstruction_runtime.py from AID-M-003.005-G001 and its signed aidefend.reconstruction-reference.v2. This example uses that same autoencoder, RGB preprocessing and float32 per-image mean squared error; it does not substitute an unrelated inverter/generator. A different architecture requires compatible measured build and scoring adapters, not invented bindings. The protected deployment profile binds the policy and reference paths, digests, signatures and verification keys, artifact bounds, timeout and positive anomaly threshold. None comes from a media request or a caller-supplied trusted flag.

Verify captured bytes, then score
from __future__ import annotations

# File: media_reconstruction_detector.py; use the exact M-003.005 shared module.
from reconstruction_runtime import (
    finite_number, load_runtime, preprocess_image, reconstruction_errors,
    sha256, strict_json, validate_reference, verified_bytes,
)


def score_reconstruction(*, image_bytes, expected_image_sha256, runtime,
                         reference_raw, z_score_threshold, now=None):
    # runtime and reference_raw are source-owned, verified loader outputs.
    value = strict_json(reference_raw)
    validate_reference(runtime, value, now=now)  # includes expiry on every use
    if sha256(image_bytes) != expected_image_sha256:
        raise ValueError("media differs from the authenticated ingress binding")
    z = finite_number(z_score_threshold, "z_score_threshold")
    if z <= 0:
        raise ValueError("the signed detector threshold must be positive")
    image = preprocess_image(image_bytes, runtime.preprocessing)
    mse = float(reconstruction_errors(runtime, image)[0])
    threshold = finite_number(value["mean_mse"] + z * value["std_mse"], "threshold")
    return {
        "schema_version": "aidefend.reconstruction_finding.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if mse > threshold else "NO_FINDING",
        "image_sha256": expected_image_sha256,
        "baseline_sha256": sha256(reference_raw),
        "baseline_version": value["reference_version"],
        "policy_version": runtime.policy["policy_version"],
        "model_sha256": value["model_weights_sha256"],
        "preprocessing_sha256": value["preprocessing_sha256"],
        "media_domain": value["media_domain"],
        "mse": mse, "threshold": threshold,
    }


def detect_image(*, image_bytes, expected_image_sha256, deployment, now=None):
    # deployment comes from the protected, signature-verified release profile.
    # The client cannot supply paths, verification keys, clock, digests or threshold.
    common = {"maximum_bytes": deployment["maximum_artifact_bytes"],
              "command_timeout": deployment["command_timeout"]}
    policy_raw = verified_bytes(deployment["policy_path"], deployment["policy_signature"],
                                deployment["policy_verify_key"], deployment["policy_sha256"], **common)
    runtime = load_runtime(policy_raw, maximum_artifact_bytes=deployment["maximum_artifact_bytes"])
    reference_raw = verified_bytes(deployment["reference_path"], deployment["reference_signature"],
                                   deployment["reference_verify_key"], deployment["reference_sha256"], **common)
    return score_reconstruction(image_bytes=image_bytes, expected_image_sha256=expected_image_sha256,
                                runtime=runtime, reference_raw=reference_raw,
                                z_score_threshold=deployment["z_score_threshold"], now=now)
Verify the integration and report failures honestly

The public entry point detect_image verifies both signatures and deployment digests over private captured bytes before loading the model and reference. It re-hashes the exact model/configuration, shared pipeline and preprocessing; validate_reference checks their bindings, complete measured population, error definition and policy validity on each score, including when a verified artifact has been cached. Only the trusted service clock supplies time. Expired/future references, missing artifacts or incomplete measurements are INSUFFICIENT_DATA; malformed, wrong-bound, unreadable or non-finite data and verification/model failures are ERROR. The caller records that state; exceptions must never be replaced by NO_FINDING.

Replay the actual builder output through this scorer with clean and anomalous images, and reject wrong preprocessing/model/pipeline, expired/future, incomplete, altered, malformed, oversized and decompression-bomb fixtures. Independently read the emitted finding by image/reference digest and recompute it with the same release. This result is an anomaly signal, not proof of a deepfake or media authenticity; blocking belongs to the applicable admission control.

AID-D-001.002-G006

Cluster recent reconstruction-anomaly signals to identify coordinated bursts of suspicious synthetic media.
Concept:

A single anomalous image may be benign corruption, but a burst of similar high-error assets usually indicates a probing or deepfake campaign. This Detect-side control groups recent anomaly events so investigators can see campaign shape, not just isolated alerts.

Step 1: Buffer recent anomaly events with stable features
from __future__ import annotations

# File: detection/media_anomaly_burst.py

import hmac as _aidefend_policy_hmac
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import re as _aidefend_policy_re
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_write_private(path, payload):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_WRONLY
        | _aidefend_policy_os.O_CREAT
        | _aidefend_policy_os.O_EXCL,
        0o600,
    )
    with _aidefend_policy_os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        _aidefend_policy_os.fsync(stream.fileno())
    _aidefend_policy_os.chmod(path, 0o400)


def _aidefend_strict_json(raw):
    def reject_duplicate_keys(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate JSON key: " + key)
            value[key] = item
        return value

    return _aidefend_policy_json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    if policy_path == signature_path:
        raise RuntimeError("policy and signature paths must differ")
    policy_raw = policy_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = _aidefend_policy_os.environ[key_env]
    remote_key = bool(_aidefend_policy_re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*:(?://)?", key_reference
    )) and not key_reference.startswith("-----BEGIN ")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else _AidefendPolicyPath(key_reference).read_bytes()
    )
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    with _aidefend_policy_tempfile.TemporaryDirectory(
        prefix="aidefend-signed-policy-"
    ) as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        policy_snapshot = root / "policy.json"
        signature_snapshot = root / "policy.sig"
        _aidefend_write_private(policy_snapshot, policy_raw)
        _aidefend_write_private(signature_snapshot, signature_raw)
        captured = [(policy_snapshot, policy_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            _aidefend_policy_subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(policy_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=verify_timeout,
            )
            if any(
                not _aidefend_policy_hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            policy = _aidefend_strict_json(policy_raw)
        finally:
            for path, _ in captured:
                _aidefend_policy_os.chmod(path, 0o600)
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy


from collections import deque
from dataclasses import dataclass
import time

import numpy as np
from sklearn.cluster import DBSCAN

CORRELATION_POLICY = _aidefend_load_signed_policy(
    path_env="D001002_CORRELATION_POLICY_PATH",
    signature_env="D001002_CORRELATION_POLICY_SIGNATURE",
    key_env="D001002_CORRELATION_POLICY_VERIFY_KEY",
    schema_version="aidefend.media-anomaly-correlation-policy.v1",
)
WINDOW_SECONDS = CORRELATION_POLICY.get("window_seconds")
MINIMUM_EVENTS = CORRELATION_POLICY.get("minimum_events")
DBSCAN_EPSILON = CORRELATION_POLICY.get("dbscan_epsilon")
DBSCAN_MINIMUM_SAMPLES = CORRELATION_POLICY.get("dbscan_minimum_samples")
if (
    any(isinstance(value, bool) or not isinstance(value, int) or value < 1
        for value in (WINDOW_SECONDS, MINIMUM_EVENTS, DBSCAN_MINIMUM_SAMPLES))
    or isinstance(DBSCAN_EPSILON, bool)
    or not isinstance(DBSCAN_EPSILON, (int, float))
    or float(DBSCAN_EPSILON) <= 0
):
    raise RuntimeError("signed media-correlation bounds are invalid")
EVENTS: deque = deque()


@dataclass(frozen=True)
class MediaAnomalyEvent:
    ts: int
    mse: float
    face_score: float
    embedding_distance: float
    source_key: str


def add_event(event: MediaAnomalyEvent) -> None:
    EVENTS.append(event)
    cutoff = int(time.time()) - WINDOW_SECONDS
    while EVENTS and EVENTS[0].ts < cutoff:
        EVENTS.popleft()
Step 2: Cluster the recent anomaly stream and raise campaign findings
# File: detection/media_anomaly_burst_query.py
import numpy as np
from sklearn.cluster import DBSCAN

from detection.media_anomaly_burst import (
    DBSCAN_EPSILON,
    DBSCAN_MINIMUM_SAMPLES,
    EVENTS,
    MINIMUM_EVENTS,
)

def find_bursts() -> list[dict[str, object]]:
    if len(EVENTS) < MINIMUM_EVENTS:
        return []

    feature_matrix = np.array([
        [event.mse, event.face_score, event.embedding_distance]
        for event in EVENTS
    ])
    labels = DBSCAN(eps=float(DBSCAN_EPSILON), min_samples=DBSCAN_MINIMUM_SAMPLES).fit(feature_matrix).labels_

    findings = []
    for label in sorted(set(labels)):
        if label == -1:
            continue
        members = [event for event, member_label in zip(EVENTS, labels) if member_label == label]
        findings.append({
            'cluster_id': int(label),
            'event_count': len(members),
            'source_keys': sorted({event.source_key for event in members}),
            'mean_mse': round(float(np.mean([event.mse for event in members])), 6),
        })
    return findings

Action: Feed reconstruction-error findings into a rolling burst-analysis job. Alert when DBSCAN or an equivalent density method identifies small, repeated clusters of anomalous media from the same source family or campaign window.

Verify safely

An independent verifier must replay the exact immutable event window with a separately pinned clustering implementation and compare membership, cluster labels, and burst findings, read the stored objects back, verify signatures and digests, and compare the canonical result digest.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-001.002: Synthetic Media & Deepfake Forensics | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service