┌─────────────────────────────┐
   이미지  ──▶   │   FastAPI POST /verify       │
                │                              │
                │   1. 업로드 검증              │
                │   2. ┌──────────────────┐    │
                │      │ C2PA 매니페스트    │    │  출처 신호
                │      │ c2pa-python       │    │
                │      └──────────────────┘    │
                │   3. ┌──────────────────┐    │
                │      │ 분류기 API         │    │  통계 신호
                │      │ 호스팅 탐지기      │    │
                │      └──────────────────┘    │
                │   4. 판정 결합               │
                └─────────────────────────────┘
                              │
                              ▼
                   JSON 판정 + 신뢰도

pip install fastapi "uvicorn[standard]" python-multipart httpx c2pa-python

# provenance.py
import json
import c2pa


def read_provenance(image_path: str) -> dict:
    """
    이미지에서 C2PA 매니페스트를 읽고 검증한다.
    결과는 서비스 내부에서 사용하기 쉬운 dict로 정규화한다.
    """
    try:
        with c2pa.Reader(image_path) as reader:
            manifest_store = json.loads(reader.json())

    except c2pa.C2paError as err:
        # 대부분의 이미지에서 예상되는 정상 케이스
        if str(err).startswith("ManifestNotFound"):
            return {
                "has_manifest": False,
                "validation": "none",
                "detail": "No C2PA manifest present in this image.",
            }

        # C2PA 데이터가 있지만 파싱 또는 검증에 실패한 경우
        return {
            "has_manifest": True,
            "validation": "error",
            "detail": f"Could not parse manifest: {err}",
        }

    active_label = manifest_store.get("active_manifest")
    manifests = manifest_store.get("manifests", {})
    active = manifests.get(active_label, {})

    # validation_status는 검증 문제가 있을 때 채워진다.
    validation_status = manifest_store.get("validation_status", [])
    validation = "valid" if not validation_status else "invalid"

    claim_generator = active.get("claim_generator", "unknown")
    signature_issuer = active.get("signature_info", {}).get("issuer", "unknown")

    return {
        "has_manifest": True,
        "validation": validation,
        "claim_generator": claim_generator,
        "signature_issuer": signature_issuer,
        "validation_status": validation_status,
        "detail": "Manifest read successfully.",
    }

https://api.sightengine.com/1.0/check.json

# classifier.py
import httpx

SIGHTENGINE_URL = "https://api.sightengine.com/1.0/check.json"


async def classify_image(
    image_bytes: bytes,
    filename: str,
    api_user: str,
    api_secret: str,
    timeout_seconds: float = 8.0,
) -> dict:
    """
    이미지를 호스팅 탐지기에 전송한다.
    성공 시 AI 생성 점수를 반환하고,
    실패 시 available: False와 reason을 반환한다.
    """
    data = {
        "models": "genai",
        "api_user": api_user,
        "api_secret": api_secret,
    }
    files = {"media": (filename, image_bytes)}

    try:
        async with httpx.AsyncClient(timeout=timeout_seconds) as client:
            response = await client.post(SIGHTENGINE_URL, data=data, files=files)
            response.raise_for_status()
            payload = response.json()

    except httpx.TimeoutException:
        return {"available": False, "reason": "classifier_timeout"}

    except httpx.HTTPStatusError as err:
        return {
            "available": False,
            "reason": f"classifier_http_{err.response.status_code}",
        }

    except httpx.HTTPError as err:
        return {"available": False, "reason": f"classifier_error: {err}"}

    if payload.get("status") != "success":
        return {
            "available": False,
            "reason": payload.get("error", {}).get("message", "unknown_error"),
        }

    ai_score = payload.get("type", {}).get("ai_generated")
    if ai_score is None:
        return {"available": False, "reason": "missing_score_in_response"}

    return {"available": True, "ai_score": float(ai_score)}

POST /verify
Content-Type: multipart/form-data

{
  "verdict": "likely_ai",
  "confidence": 0.86,
  "signals": {
    "provenance": {
      "has_manifest": true,
      "validation": "valid",
      "claim_generator": "SomeImageTool/2.1",
      "signature_issuer": "Some Issuing CA"
    },
    "classifier": {
      "available": true,
      "ai_score": 0.91
    }
  },
  "explanation": "유효한 C2PA 매니페스트가 AI 이미지 도구를 명시하며, 분류기가 해당 이미지를 AI 생성으로 추정합니다.",
  "checked_at": "2026-05-21T09:30:00Z"
}

components:
  schemas:
    VerifyResponse:
      type: object
      required: [verdict, confidence, signals, checked_at]
      properties:
        verdict:
          type: string
          enum: [likely_authentic, likely_ai, uncertain]
        confidence:
          type: number
          format: float
          minimum: 0
          maximum: 1
        signals:
          type: object
          properties:
            provenance:
              type: object
              properties:
                has_manifest:
                  type: boolean
                validation:
                  type: string
                  enum: [valid, invalid, error, none]
                claim_generator:
                  type: string
                signature_issuer:
                  type: string
            classifier:
              type: object
              properties:
                available:
                  type: boolean
                ai_score:
                  type: number
                  format: float
        explanation:
          type: string
        checked_at:
          type: string
          format: date-time

# verdict.py


def combine_signals(provenance: dict, classifier: dict) -> dict:
    """출처 신호와 분류기 신호를 하나의 판정으로 결합한다."""
    has_manifest = provenance.get("has_manifest", False)
    validation = provenance.get("validation", "none")
    generator = (provenance.get("claim_generator") or "").lower()

    classifier_ok = classifier.get("available", False)
    ai_score = classifier.get("ai_score")

    ai_keywords = (
        "firefly",
        "dall-e",
        "dalle",
        "midjourney",
        "stable",
        "gpt",
        "gemini",
        "imagen",
        "generat",
    )
    generator_looks_ai = any(k in generator for k in ai_keywords)

    # 1. 유효한 매니페스트가 AI 생성 도구를 명시하는 경우
    if has_manifest and validation == "valid" and generator_looks_ai:
        return _verdict(
            "likely_ai",
            0.95,
            "A valid C2PA manifest names an AI image tool.",
        )

    # 2. 유효한 매니페스트가 카메라 또는 비 AI 도구를 명시하는 경우
    if has_manifest and validation == "valid" and not generator_looks_ai:
        if classifier_ok and ai_score is not None and ai_score > 0.85:
            return _verdict(
                "uncertain",
                0.55,
                "Manifest looks authentic but the classifier disagrees; signals conflict.",
            )

        return _verdict(
            "likely_authentic",
            0.9,
            "A valid C2PA manifest from a non-AI tool is present.",
        )

    # 3. 매니페스트가 있지만 검증 실패한 경우
    if has_manifest and validation in ("invalid", "error"):
        return _verdict(
            "uncertain",
            0.6,
            "The image carries a C2PA manifest that failed validation; its claimed history is unverified.",
        )

    # 4. 매니페스트가 없으면 분류기에 의존
    if classifier_ok and ai_score is not None:
        if ai_score >= 0.7:
            return _verdict(
                "likely_ai",
                round(ai_score, 2),
                "No provenance data; the classifier scored the image as likely AI-generated.",
            )

        if ai_score <= 0.3:
            return _verdict(
                "likely_authentic",
                round(1 - ai_score, 2),
                "No provenance data; the classifier scored the image as likely authentic.",
            )

        return _verdict(
            "uncertain",
            0.5,
            "No provenance data and the classifier score is inconclusive.",
        )

    # 5. 두 신호 모두 없는 경우
    return _verdict(
        "uncertain",
        0.0,
        "No provenance data and the classifier was unavailable.",
    )


def _verdict(verdict: str, confidence: float, explanation: str) -> dict:
    return {
        "verdict": verdict,
        "confidence": confidence,
        "explanation": explanation,
    }

# main.py
import os
import tempfile
from datetime import datetime, timezone

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse

from provenance import read_provenance
from classifier import classify_image
from verdict import combine_signals

app = FastAPI(title="AI Image Detector API", version="1.0.0")

ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}
MAX_BYTES = 12 * 1024 * 1024  # 12 MB

SIGHTENGINE_USER = os.environ.get("SIGHTENGINE_API_USER", "")
SIGHTENGINE_SECRET = os.environ.get("SIGHTENGINE_API_SECRET", "")


@app.post("/verify")
async def verify(image: UploadFile = File(...)):
    # 1. 업로드 검증
    if image.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail=(
                f"Unsupported type {image.content_type}. "
                "Send JPEG, PNG, or WebP."
            ),
        )

    image_bytes = await image.read()

    if len(image_bytes) == 0:
        raise HTTPException(status_code=400, detail="Empty file.")

    if len(image_bytes) > MAX_BYTES:
        raise HTTPException(status_code=413, detail="File exceeds 12 MB limit.")

    # 2. C2PA 출처 신호
    # c2pa.Reader는 파일 경로를 필요로 하므로 임시 파일을 사용한다.
    suffix = os.path.splitext(image.filename or "")[1] or ".img"

    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
        tmp.write(image_bytes)
        tmp_path = tmp.name

    try:
        provenance = read_provenance(tmp_path)
    finally:
        os.unlink(tmp_path)

    # 3. 분류기 신호
    # 분류기 실패는 예외가 아니라 available: False로 처리한다.
    if SIGHTENGINE_USER and SIGHTENGINE_SECRET:
        classifier = await classify_image(
            image_bytes=image_bytes,
            filename=image.filename or "upload",
            api_user=SIGHTENGINE_USER,
            api_secret=SIGHTENGINE_SECRET,
        )
    else:
        classifier = {
            "available": False,
            "reason": "classifier_not_configured",
        }

    # 4. 신호 결합
    result = combine_signals(provenance, classifier)

    return JSONResponse(
        {
            "verdict": result["verdict"],
            "confidence": result["confidence"],
            "signals": {
                "provenance": {
                    k: provenance.get(k)
                    for k in (
                        "has_manifest",
                        "validation",
                        "claim_generator",
                        "signature_issuer",
                    )
                },
                "classifier": {
                    "available": classifier.get("available", False),
                    "ai_score": classifier.get("ai_score"),
                },
            },
            "explanation": result["explanation"],
            "checked_at": datetime.now(timezone.utc).isoformat(),
        }
    )

uvicorn main:app --reload

http://127.0.0.1:8000/verify

export SIGHTENGINE_API_USER="your_api_user"
export SIGHTENGINE_API_SECRET="your_api_secret"

from PIL import Image
from io import BytesIO

def validate_image_bytes(image_bytes: bytes) -> None:
    try:
        with Image.open(BytesIO(image_bytes)) as img:
            img.verify()
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid image file.")

pip install pillow

{
  "available": false,
  "reason": "classifier_timeout"
}