synthid_gate.py

"""Optional SynthID (Google) watermark check via the Content Detection API.

Google's *image* SynthID detection is NOT open source, and the Content Detection
API is in limited preview (waitlist). So this gate is OFF by default and is a pure
no-op — the backend behaves exactly as before. When you are granted access, set:

    SYNTHID_API_URL   the REST endpoint Google gives you on approval
    SYNTHID_API_KEY   your API key / bearer token
    SYNTHID_API_FIELD (optional) multipart field name for the image (default "image")

...and it activates. SynthID reads an invisible *pixel* watermark, so unlike our
byte/metadata provenance_gate it survives screenshots / re-encoding / messenger
re-saves — that is the whole point of wiring it in.

NOTE: the request/response shape below is a best-effort guess for the preview API.
Once you have the real contract, only `_parse_detected()` and the POST body likely
need a one-line tweak.
"""
from __future__ import annotations

import os

import requests

_URL = os.getenv("SYNTHID_API_URL", "").strip()
_KEY = os.getenv("SYNTHID_API_KEY", "").strip()
_FIELD = os.getenv("SYNTHID_API_FIELD", "image").strip() or "image"
_TIMEOUT = float(os.getenv("SYNTHID_API_TIMEOUT", "8"))


def enabled() -> bool:
    """True only when both the endpoint and key are configured."""
    return bool(_URL and _KEY)


def _parse_detected(data: dict) -> bool:
    """Best-effort: did the API say this image carries a watermark? Adjust to the
    real preview contract when you have the docs."""
    if not isinstance(data, dict):
        return False
    for key in ("watermark_detected", "synthid_detected", "is_ai", "detected"):
        if bool(data.get(key)):
            return True
    verdict = str(data.get("verdict") or data.get("label") or "").lower()
    if verdict in {"ai", "ai-generated", "watermarked", "synthid", "synthetic"}:
        return True
    # some APIs return a confidence; treat >= 0.5 as detected
    conf = data.get("confidence") or data.get("score")
    return isinstance(conf, (int, float)) and float(conf) >= 0.5


def check_image(image_bytes: bytes, content_type: str = "image/jpeg") -> dict:
    """Returns {'signals': [...], 'details': {...}}. Empty signals = not watermarked
    OR not configured OR call failed (never raises)."""
    if not enabled():
        return {"signals": [], "details": {"synthid": "disabled"}}
    try:
        resp = requests.post(
            _URL,
            headers={"Authorization": f"Bearer {_KEY}"},
            files={_FIELD: ("upload", image_bytes, content_type)},
            timeout=_TIMEOUT,
        )
        if resp.status_code != 200:
            return {"signals": [], "details": {"synthid_http": resp.status_code}}
        data = resp.json()
        if _parse_detected(data):
            return {"signals": ["Google/SynthID-API"], "details": {"synthid": data}}
        return {"signals": [], "details": {"synthid": data}}
    except Exception as exc:  # noqa: BLE001 — provenance must never break analysis
        return {"signals": [], "details": {"synthid_error": str(exc)[:160]}}