"""C2PA + IPTC + EXIF provenance detection service. Checks media files for AI-generation provenance signals embedded in C2PA manifests, EXIF metadata, and IPTC fields. Returns a probability score indicating likelihood of AI generation based on metadata evidence alone. Unlike detection models that analyze visual/audio artifacts, this service inspects only embedded metadata and watermarks. """ import base64 import logging import os import sys import tempfile import time from typing import Any, Dict, Optional, Set import uvicorn from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel # ── Logging ──────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger(__name__) # ── Config ───────────────────────────────────────────────────────────────── MODEL_NAME = "c2pa_checker" MODEL_PORT = int(os.environ.get("MODEL_PORT", "9001")) _PRODUCTION = os.environ.get("PRODUCTION", "false").lower() == "true" # Known AI generator strings to match in C2PA manifests and EXIF metadata. _AI_GENERATORS: Set[str] = { "adobe firefly", "dall-e", "dall\u00b7e", "openai", "midjourney", "stable diffusion", "stability ai", "google gemini", "imagen", "microsoft designer", "copilot", "meta ai", "leonardo ai", "runway", "sora", "kling", "flux", "ideogram", "veo", "google generative ai", "trainedalgorithmicmedia", } # ── Pydantic models ─────────────────────────────────────────────────────── class PredictInput(BaseModel): """Request body for /predict.""" image_data: Optional[str] = None audio_data: Optional[str] = None video_data: Optional[str] = None threshold: float = 0.5 # ── Detection helpers ────────────────────────────────────────────────────── def _check_c2pa(file_path: str) -> tuple: """Check file for C2PA manifest with AI generator assertions. Verifies the cryptographic signature chain. Unsigned or invalid manifests are downweighted because they can be forged. Args: file_path: Path to the media file on disk. Returns: Tuple of (score_or_none, has_valid_c2pa). Score is 0.95 for signed AI generator, 0.60 for unsigned, None if no C2PA manifest. has_valid_c2pa indicates whether a cryptographically valid manifest exists (used to gate EXIF/IPTC trust). """ try: import c2pa reader = c2pa.Reader.try_create(file_path) if reader is None: return None, False manifest_json = reader.json() # Check signature validation status via manifest JSON # reader.is_valid is a bool property (not a method) sig_valid = True try: if hasattr(reader, "is_valid"): sig_valid = reader.is_valid # Also check validation_status in the JSON for untrusted certs import json as _json mdata = _json.loads(manifest_json) vstatus = mdata.get("validation_status", []) if vstatus: # untrusted signing credential is common for valid AI certs # only mark invalid for actual signature failures sig_codes = [v.get("code", "") for v in vstatus] has_sig_failure = any( "signature" in c and "untrusted" not in c for c in sig_codes ) if has_sig_failure: sig_valid = False logger.info("C2PA manifest has signature issues: %s", sig_codes) else: logger.info( "C2PA manifest has untrusted cert (normal for AI): %s", sig_codes, ) except Exception: pass # Search for AI generator strings manifest_lower = manifest_json.lower() for generator in _AI_GENERATORS: if generator in manifest_lower: if sig_valid: logger.info( "C2PA (verified): AI generator '%s' detected.", generator, ) return 0.95, True else: logger.info( "C2PA (UNVERIFIED): AI generator '%s' — " "signature invalid, downweighting.", generator, ) return 0.60, False # C2PA manifest exists but no AI generator logger.info("C2PA manifest found but no AI generator detected.") return None, sig_valid except Exception: return None, False def _check_exif_iptc( file_path: str, has_valid_c2pa: bool = False, ) -> Optional[float]: """Check file EXIF/IPTC metadata for AI generation indicators. IPTC/EXIF tags are trivially forgeable (one exiftool command). They are only trusted when backed by a valid C2PA manifest. Without C2PA, they receive near-neutral scores (0.55) that contribute minimal ensemble impact. Args: file_path: Path to the media file on disk. has_valid_c2pa: Whether a cryptographically valid C2PA manifest was found. Raises trust in EXIF/IPTC tags. Returns: 0.90 if backed by C2PA + DigitalSourceType AI tag, 0.85 if backed by C2PA + EXIF AI tool signature, None if IPTC/EXIF AI indicators found WITHOUT C2PA (no trust), None if no AI indicators found. """ try: import exiftool with exiftool.ExifToolHelper() as et: metadata_list = et.get_metadata(file_path) if not metadata_list: return None metadata = metadata_list[0] # Check IPTC DigitalSourceType for key, value in metadata.items(): key_lower = key.lower() if "digitalsourcetype" in key_lower: value_str = str(value).lower() if "trainedalgorithmicmedia" in value_str: if has_valid_c2pa: logger.info("IPTC AI tag (C2PA-backed): %s", value) return 0.90 else: logger.info( "IPTC AI tag (NO C2PA — low trust): %s", value, ) return None # No trust without C2PA # Check EXIF Software, Description, UserComment text_fields = [] for key, value in metadata.items(): key_lower = key.lower() if any( field in key_lower for field in ("software", "description", "usercomment") ): text_fields.append(str(value).lower()) combined_text = " ".join(text_fields) for generator in _AI_GENERATORS: if generator in combined_text: if has_valid_c2pa: logger.info( "EXIF AI signature (C2PA-backed): %s", generator, ) return 0.85 else: logger.info( "EXIF AI signature (NO C2PA — ignored): %s", generator, ) return None # No trust without C2PA return None except Exception: return None def _detect_provenance(file_bytes: bytes, data_key: str = "") -> float: """Run all provenance checks on raw file bytes. Writes bytes to a temporary file, runs C2PA and EXIF/IPTC checks, and returns the maximum signal found. Returns 0.5 (neutral) when no AI provenance is detected. Args: file_bytes: Raw bytes of the media file. data_key: The payload key used (image_data, audio_data, video_data) to determine file extension for C2PA. Returns: Float probability in [0, 1]. 0.5 means neutral (no signal). """ if not file_bytes: return 0.5 # C2PA needs a file extension to determine MIME type _EXT_MAP = { "video_data": ".mp4", "image_data": ".jpg", "audio_data": ".wav", } suffix = _EXT_MAP.get(data_key, ".bin") tmp_path = None try: with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp: tmp.write(file_bytes) tmp_path = tmp.name signals = [] c2pa_score, has_valid_c2pa = _check_c2pa(tmp_path) if c2pa_score is not None: signals.append(c2pa_score) exif_result = _check_exif_iptc(tmp_path, has_valid_c2pa) if exif_result is not None: signals.append(exif_result) if signals: return max(signals) return 0.5 except Exception as exc: logger.warning("Provenance detection error: %s", exc) return 0.5 finally: if tmp_path and os.path.exists(tmp_path): try: os.unlink(tmp_path) except OSError: pass # ── FastAPI app ──────────────────────────────────────────────────────────── app = FastAPI( title="C2PA Provenance Checker Service", description=( "Checks media files for AI-generation provenance signals in " "C2PA manifests, EXIF metadata, and IPTC fields." ), version="1.0.0", docs_url=None if _PRODUCTION else "/docs", redoc_url=None if _PRODUCTION else "/redoc", openapi_url=None if _PRODUCTION else "/openapi.json", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") async def root(): """Root endpoint with service information.""" if _PRODUCTION: return {"status": "ok"} return { "model_name": MODEL_NAME, "description": ("C2PA + EXIF/IPTC provenance checker for AI-generated media"), "modalities": ["image", "audio", "video"], } @app.get("/health") async def health(): """Health check endpoint.""" if _PRODUCTION: return {"status": "healthy"} return { "status": "healthy", "model_name": MODEL_NAME, "ready": True, } @app.post("/predict") async def predict(payload: PredictInput) -> Dict[str, Any]: """Check media for AI-generation provenance signals. Accepts image_data, audio_data, or video_data as base64-encoded file bytes. Returns a standard detection response with probability based on metadata evidence. Args: payload: Base64-encoded media and optional threshold. Returns: Dict with model name, probability, prediction, class, and inference time. """ # Find the first non-None payload key raw_b64 = None data_key = "" for key in ("image_data", "audio_data", "video_data"): value = getattr(payload, key) if value is not None: raw_b64 = value data_key = key break if raw_b64 is None: raise HTTPException( status_code=400, detail=( "No media data provided. Include one of: " "image_data, audio_data, video_data." ), ) start = time.time() try: file_bytes = base64.b64decode(raw_b64) except Exception as exc: raise HTTPException( status_code=400, detail=f"Invalid base64 data: {exc}", ) if not file_bytes: raise HTTPException( status_code=400, detail="Empty file data after base64 decode.", ) probability = _detect_provenance(file_bytes, data_key) prediction = 1 if probability > payload.threshold else 0 class_label = "fake" if prediction == 1 else "real" inference_time = time.time() - start logger.info( "Provenance check: %s (prob=%.4f, %.3fs)", class_label, probability, inference_time, ) return { "model": MODEL_NAME, "probability": float(probability), "prediction": int(prediction), "class": class_label, "inference_time": float(inference_time), } if __name__ == "__main__": logger.info("Starting C2PA checker service on port %d", MODEL_PORT) uvicorn.run(app, host="0.0.0.0", port=MODEL_PORT)