147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
"""FastAPI endpoints for enrollment, verification, and user listing."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from sqlalchemy import select
|
|
|
|
from app.db import SessionLocal, SignatureTemplate, initialize_database
|
|
from sigverify.decision import Thresholds
|
|
from sigverify.io.schema import Signature
|
|
from sigverify.matching.models import DTWVerifier
|
|
from sigverify.preprocessing import preprocess_signature
|
|
from sigverify.security import AttemptLockout, TemplateCipher
|
|
|
|
app = FastAPI(title="Signature Verification Demo", version="0.1.0")
|
|
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
|
initialize_database()
|
|
LOCKOUT = AttemptLockout(max_failures=5, lockout_seconds=60)
|
|
THRESHOLDS = Thresholds(global_threshold=float(os.environ.get("SIGVERIFY_THRESHOLD", "1.0")))
|
|
|
|
|
|
class CapturePayload(BaseModel):
|
|
"""JSON body accepted by enrollment and verification endpoints."""
|
|
|
|
user_id: str = Field(min_length=1, max_length=200)
|
|
t: list[float] = Field(min_length=1)
|
|
x: list[float] = Field(min_length=1)
|
|
y: list[float] = Field(min_length=1)
|
|
p: list[float] | None = None
|
|
area: list[float] | None = None
|
|
pen_down: list[bool] | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_lengths(self) -> "CapturePayload":
|
|
length = len(self.t)
|
|
for name in ("x", "y", "p", "area", "pen_down"):
|
|
values = getattr(self, name)
|
|
if values is not None and len(values) != length:
|
|
raise ValueError(f"{name} must have the same number of points as t")
|
|
return self
|
|
|
|
def to_signature(self, label: str) -> Signature:
|
|
return Signature(
|
|
user_id=self.user_id,
|
|
label=label,
|
|
t=self.t,
|
|
x=self.x,
|
|
y=self.y,
|
|
p=self.p,
|
|
area=self.area,
|
|
pen_down=self.pen_down,
|
|
)
|
|
|
|
|
|
def _cipher() -> TemplateCipher:
|
|
try:
|
|
return TemplateCipher()
|
|
except RuntimeError as error:
|
|
raise HTTPException(status_code=503, detail=str(error)) from error
|
|
|
|
|
|
def _prepared(payload: CapturePayload, label: str) -> Signature:
|
|
try:
|
|
return preprocess_signature(payload.to_signature(label))
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=422, detail=str(error)) from error
|
|
|
|
|
|
@app.get("/")
|
|
def home():
|
|
"""Serve the browser signing page."""
|
|
from fastapi.responses import FileResponse
|
|
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.post("/enroll")
|
|
def enroll(payload: CapturePayload) -> dict[str, str]:
|
|
"""Add one genuine enrollment template for a user."""
|
|
cipher = _cipher()
|
|
signature = _prepared(payload, "genuine")
|
|
with SessionLocal() as session:
|
|
session.add(
|
|
SignatureTemplate(
|
|
user_id=signature.user_id,
|
|
encrypted_payload=cipher.encrypt(signature),
|
|
)
|
|
)
|
|
session.commit()
|
|
return {"status": "enrolled", "user_id": signature.user_id}
|
|
|
|
|
|
@app.post("/verify")
|
|
def verify(payload: CapturePayload) -> dict[str, float | bool | str]:
|
|
"""Score a capture against all encrypted samples enrolled for its user."""
|
|
user_id = payload.user_id
|
|
remaining = LOCKOUT.remaining_lockout(user_id)
|
|
if remaining > 0:
|
|
raise HTTPException(status_code=429, detail=f"locked; retry in {remaining:.1f} seconds")
|
|
|
|
cipher = _cipher()
|
|
with SessionLocal() as session:
|
|
records = session.scalars(
|
|
select(SignatureTemplate).where(SignatureTemplate.user_id == user_id)
|
|
).all()
|
|
if not records:
|
|
raise HTTPException(status_code=404, detail="no enrollment templates for this user")
|
|
try:
|
|
templates = [cipher.decrypt(record.encrypted_payload) for record in records]
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=503, detail=str(error)) from error
|
|
|
|
verifier = DTWVerifier()
|
|
verifier.enroll(templates)
|
|
score = verifier.score(_prepared(payload, "query"))
|
|
accepted = THRESHOLDS.accept(user_id, score)
|
|
if accepted:
|
|
LOCKOUT.record_success(user_id)
|
|
else:
|
|
LOCKOUT.record_failure(user_id)
|
|
return {
|
|
"user_id": user_id,
|
|
"score": score,
|
|
"threshold": THRESHOLDS.threshold_for(user_id),
|
|
"accepted": accepted,
|
|
}
|
|
|
|
|
|
@app.get("/users")
|
|
def users() -> dict[str, list[dict[str, str | int]]]:
|
|
"""List enrolled identities and their template counts."""
|
|
with SessionLocal() as session:
|
|
records = session.scalars(select(SignatureTemplate)).all()
|
|
counts: dict[str, int] = {}
|
|
for record in records:
|
|
counts[record.user_id] = counts.get(record.user_id, 0) + 1
|
|
return {
|
|
"users": [
|
|
{"user_id": user_id, "template_count": count}
|
|
for user_id, count in sorted(counts.items())
|
|
]
|
|
}
|