Spaces:

Veridicate
/

scanner

Running

scanner / src /deepfake_scanner /api /v1.py

github-actions[bot]

Deploy from GitHub 1af737237f5d10dfa9cf12849d98f47c3b30ce37 (filtered: code + Dockerfile + README + NOTICES only)

a2f55b2 about 2 months ago

7.77 kB

"""
v1 API.
Endpoints
---------
GET  /health                  → liveness probe
GET  /v1/info                 → model + config metadata
POST /v1/scan/image           → scan an uploaded image
Privacy
-------
Image bytes live in-memory only for the duration of the request. They are
never written to disk or to blob storage in Stage 1. After the response is
returned the bytes go out of scope and are garbage-collected.
"""
from __future__ import annotations

import logging
import time
import uuid

from fastapi import FastAPI, File, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from ..config import settings
from ..detectors.ensemble import _DETECTORS, run_ensemble
from ..preprocess import InvalidImageError, decode
from ..provenance import check_c2pa
from ..verdict import decide
from ..storage import record_scan
from ..storage.db import ScanRecord
from .schemas import (
    DetectorSignal,
    ErrorResponse,
    Probabilities,
    Provenance,
    ScanResponse,
)

log = logging.getLogger(__name__)


def _detectors_for_info():
    """Live detector instances, for reporting real loaded state."""
    return _DETECTORS


_ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.api_title,
        version=settings.api_version,
        description=(
            "Detect AI-generated and manipulated images. "
            "Visitor uploads are processed in-memory and never stored."
        ),
    )

    # Browser preflight allowlist for the static-HTML frontend. Credentialed
    # requests are off (no cookies); the upload is the only state we accept.
    app.add_middleware(
        CORSMiddleware,
        allow_origin_regex=settings.cors_allow_origin_regex,
        allow_methods=["GET", "POST", "OPTIONS"],
        allow_headers=["*"],
        allow_credentials=False,
    )

    # HEAD is accepted on read-only endpoints so uptime monitors and CDN
    # validators can ping cheaply without a body. Per RFC 7231, any URL
    # that accepts GET should also accept HEAD.
    @app.api_route("/health", methods=["GET", "HEAD"])
    def health() -> dict:
        # service_version here so a single curl identifies the running build.
        return {"status": "ok", "service_version": settings.service_version}

    @app.api_route("/v1/info", methods=["GET", "HEAD"])
    def info() -> dict:
        # head_loaded reports what is ACTUALLY serving, not what is configured.
        # model_version is a config string and stays the same whether or not the
        # weights loaded - which is exactly how a random head can look healthy.
        head_loaded = None
        for det in _detectors_for_info():
            if getattr(det, "name", None) == "clip_classifier":
                head_loaded = bool(getattr(det, "is_trained", False))
        return {
            "api_version": settings.api_version,
            "service_version": settings.service_version,
            "model_version": settings.model_version,
            "head_loaded": head_loaded,
            "detectors": {
                "clip_classifier": settings.enable_clip_detector,
                "frequency_artifacts": settings.enable_frequency_detector,
                "face_swap": settings.enable_face_swap_detector,
            },
            "c2pa_check_enabled": settings.enable_c2pa_check,
            "max_upload_bytes": settings.max_upload_bytes,
            "store_uploads": settings.store_uploads,
        }

    @app.post(
        "/v1/scan/image",
        response_model=ScanResponse,
        responses={
            413: {"model": ErrorResponse},
            415: {"model": ErrorResponse},
            422: {"model": ErrorResponse},
        },
    )
    async def scan_image(file: UploadFile = File(...)) -> ScanResponse:
        if file.content_type not in _ALLOWED_TYPES:
            raise HTTPException(
                status_code=415,
                detail=(
                    f"Unsupported media type '{file.content_type}'. "
                    "Use JPEG, PNG, or WebP."
                ),
            )

        image_bytes = await file.read()

        if len(image_bytes) > settings.max_upload_bytes:
            raise HTTPException(
                status_code=413,
                detail=(
                    f"File exceeds the {settings.max_upload_bytes // (1024*1024)} MB limit."
                ),
            )

        t0 = time.perf_counter()

        # 1. Provenance check (advisory — does not yet override the model).
        provenance_result = check_c2pa(image_bytes)

        # 2. Decode image.
        try:
            image = decode(image_bytes)
        except InvalidImageError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc

        # 3. Run detector ensemble.
        ensemble_out = run_ensemble(image)

        # 3b. Let validated provenance override the model. One-directional:
        # it can only push toward `ai_generated`, never toward `authentic`.
        # See verdict.py for why a camera signature is not trusted in reverse.
        decision = decide(
            ensemble_out.verdict, ensemble_out.confidence, provenance_result
        )

        latency_ms = (time.perf_counter() - t0) * 1000.0
        scan_id = f"scn_{uuid.uuid4().hex[:24]}"

        response = ScanResponse(
            verdict=decision.verdict,
            confidence=round(decision.confidence, 4),
            verdict_source=decision.source,
            probabilities=Probabilities(
                authentic=round(ensemble_out.probabilities.authentic, 4),
                ai_generated=round(ensemble_out.probabilities.ai_generated, 4),
                deepfake=round(ensemble_out.probabilities.deepfake, 4),
                edited=round(ensemble_out.probabilities.edited, 4),
            ),
            signals=(
                ensemble_out.signals
                + [DetectorSignal(name="c2pa_provenance", score=1.0,
                                  notes=decision.reason)]
                if decision.source == "provenance"
                else ensemble_out.signals
            ),
            provenance=Provenance(
                c2pa_present=provenance_result.present,
                c2pa_valid=provenance_result.valid,
                issuer=provenance_result.issuer,
                claim_generator=provenance_result.claim_generator,
                ai_declared=provenance_result.ai_declared,
            ),
            model_version=settings.model_version,
            service_version=settings.service_version,
            scan_id=scan_id,
            latency_ms=round(latency_ms, 1),
        )

        # 4. Persist METADATA only — never the image.
        record_scan(
            ScanRecord(
                scan_id=scan_id,
                verdict=response.verdict,
                confidence=response.confidence,
                model_version=response.model_version,
                latency_ms=response.latency_ms,
                c2pa_present=provenance_result.present,
            )
        )

        # 5. Drop the image bytes ASAP. (Local var goes out of scope when the
        # function returns; the explicit del documents the privacy intent.)
        del image_bytes

        return response

    @app.exception_handler(HTTPException)
    async def http_exception_handler(_, exc: HTTPException) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content=ErrorResponse(
                error=f"HTTP {exc.status_code}",
                detail=str(exc.detail),
            ).model_dump(),
        )

    return app


# Default app instance for `uvicorn deepfake_scanner.api.v1:app`.
app = create_app()

	"""
	v1 API.

	Endpoints
	---------
	GET /health → liveness probe
	GET /v1/info → model + config metadata
	POST /v1/scan/image → scan an uploaded image

	Privacy
	-------
	Image bytes live in-memory only for the duration of the request. They are
	never written to disk or to blob storage in Stage 1. After the response is
	returned the bytes go out of scope and are garbage-collected.
	"""
	from __future__ import annotations

	import logging
	import time
	import uuid

	from fastapi import FastAPI, File, HTTPException, UploadFile
	from fastapi.middleware.cors import CORSMiddleware
	from fastapi.responses import JSONResponse

	from ..config import settings
	from ..detectors.ensemble import _DETECTORS, run_ensemble
	from ..preprocess import InvalidImageError, decode
	from ..provenance import check_c2pa
	from ..verdict import decide
	from ..storage import record_scan
	from ..storage.db import ScanRecord
	from .schemas import (
	DetectorSignal,
	ErrorResponse,
	Probabilities,
	Provenance,
	ScanResponse,
	)

	log = logging.getLogger(__name__)


	def _detectors_for_info():
	"""Live detector instances, for reporting real loaded state."""
	return _DETECTORS


	_ALLOWED_TYPES = {"image/jpeg", "image/png", "image/webp"}


	def create_app() -> FastAPI:
	app = FastAPI(
	title=settings.api_title,
	version=settings.api_version,
	description=(
	"Detect AI-generated and manipulated images. "
	"Visitor uploads are processed in-memory and never stored."
	),
	)

	# Browser preflight allowlist for the static-HTML frontend. Credentialed
	# requests are off (no cookies); the upload is the only state we accept.
	app.add_middleware(
	CORSMiddleware,
	allow_origin_regex=settings.cors_allow_origin_regex,
	allow_methods=["GET", "POST", "OPTIONS"],
	allow_headers=["*"],
	allow_credentials=False,
	)

	# HEAD is accepted on read-only endpoints so uptime monitors and CDN
	# validators can ping cheaply without a body. Per RFC 7231, any URL
	# that accepts GET should also accept HEAD.
	@app.api_route("/health", methods=["GET", "HEAD"])
	def health() -> dict:
	# service_version here so a single curl identifies the running build.
	return {"status": "ok", "service_version": settings.service_version}

	@app.api_route("/v1/info", methods=["GET", "HEAD"])
	def info() -> dict:
	# head_loaded reports what is ACTUALLY serving, not what is configured.
	# model_version is a config string and stays the same whether or not the
	# weights loaded - which is exactly how a random head can look healthy.
	head_loaded = None
	for det in _detectors_for_info():
	if getattr(det, "name", None) == "clip_classifier":
	head_loaded = bool(getattr(det, "is_trained", False))
	return {
	"api_version": settings.api_version,
	"service_version": settings.service_version,
	"model_version": settings.model_version,
	"head_loaded": head_loaded,
	"detectors": {
	"clip_classifier": settings.enable_clip_detector,
	"frequency_artifacts": settings.enable_frequency_detector,
	"face_swap": settings.enable_face_swap_detector,
	},
	"c2pa_check_enabled": settings.enable_c2pa_check,
	"max_upload_bytes": settings.max_upload_bytes,
	"store_uploads": settings.store_uploads,
	}

	@app.post(
	"/v1/scan/image",
	response_model=ScanResponse,
	responses={
	413: {"model": ErrorResponse},
	415: {"model": ErrorResponse},
	422: {"model": ErrorResponse},
	},
	)
	async def scan_image(file: UploadFile = File(...)) -> ScanResponse:
	if file.content_type not in _ALLOWED_TYPES:
	raise HTTPException(
	status_code=415,
	detail=(
	f"Unsupported media type '{file.content_type}'. "
	"Use JPEG, PNG, or WebP."
	),
	)

	image_bytes = await file.read()

	if len(image_bytes) > settings.max_upload_bytes:
	raise HTTPException(
	status_code=413,
	detail=(
	f"File exceeds the {settings.max_upload_bytes // (1024*1024)} MB limit."
	),
	)

	t0 = time.perf_counter()

	# 1. Provenance check (advisory — does not yet override the model).
	provenance_result = check_c2pa(image_bytes)

	# 2. Decode image.
	try:
	image = decode(image_bytes)
	except InvalidImageError as exc:
	raise HTTPException(status_code=422, detail=str(exc)) from exc

	# 3. Run detector ensemble.
	ensemble_out = run_ensemble(image)

	# 3b. Let validated provenance override the model. One-directional:
	# it can only push toward `ai_generated`, never toward `authentic`.
	# See verdict.py for why a camera signature is not trusted in reverse.
	decision = decide(
	ensemble_out.verdict, ensemble_out.confidence, provenance_result
	)

	latency_ms = (time.perf_counter() - t0) * 1000.0
	scan_id = f"scn_{uuid.uuid4().hex[:24]}"

	response = ScanResponse(
	verdict=decision.verdict,
	confidence=round(decision.confidence, 4),
	verdict_source=decision.source,
	probabilities=Probabilities(
	authentic=round(ensemble_out.probabilities.authentic, 4),
	ai_generated=round(ensemble_out.probabilities.ai_generated, 4),
	deepfake=round(ensemble_out.probabilities.deepfake, 4),
	edited=round(ensemble_out.probabilities.edited, 4),
	),
	signals=(
	ensemble_out.signals
	+ [DetectorSignal(name="c2pa_provenance", score=1.0,
	notes=decision.reason)]
	if decision.source == "provenance"
	else ensemble_out.signals
	),
	provenance=Provenance(
	c2pa_present=provenance_result.present,
	c2pa_valid=provenance_result.valid,
	issuer=provenance_result.issuer,
	claim_generator=provenance_result.claim_generator,
	ai_declared=provenance_result.ai_declared,
	),
	model_version=settings.model_version,
	service_version=settings.service_version,
	scan_id=scan_id,
	latency_ms=round(latency_ms, 1),
	)

	# 4. Persist METADATA only — never the image.
	record_scan(
	ScanRecord(
	scan_id=scan_id,
	verdict=response.verdict,
	confidence=response.confidence,
	model_version=response.model_version,
	latency_ms=response.latency_ms,
	c2pa_present=provenance_result.present,
	)
	)

	# 5. Drop the image bytes ASAP. (Local var goes out of scope when the
	# function returns; the explicit del documents the privacy intent.)
	del image_bytes

	return response

	@app.exception_handler(HTTPException)
	async def http_exception_handler(_, exc: HTTPException) -> JSONResponse:
	return JSONResponse(
	status_code=exc.status_code,
	content=ErrorResponse(
	error=f"HTTP {exc.status_code}",
	detail=str(exc.detail),
	).model_dump(),
	)

	return app


	# Default app instance for `uvicorn deepfake_scanner.api.v1:app`.
	app = create_app()