commit f9265f2d4dfe9146e0f1b6fc95d70384f9d20801 Author: Ján Malinovský Date: Mon Oct 19 08:11:33 2026 +0200 Prvy diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5beb22e --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +.venv/ +__pycache__/ +.pytest_cache/ +*.py[cod] +*.db +*.sqlite3 +.env +data/raw/* +!data/raw/*/ +!data/raw/*/.gitkeep +data/processed/* +!data/processed/.gitkeep +results/* +!results/.gitkeep +*.png +*.pdf +*.csv +!src/**/*.csv diff --git a/MOBISIG.ZIP b/MOBISIG.ZIP new file mode 100644 index 0000000..d53b502 Binary files /dev/null and b/MOBISIG.ZIP differ diff --git a/README.md b/README.md new file mode 100644 index 0000000..c72e4ab --- /dev/null +++ b/README.md @@ -0,0 +1,78 @@ +# Signature verification baseline + +A small research starter project for online handwritten-signature verification. +It includes common-format CSV/JSON loading, preprocessing, feature extraction, +DTW matching, threshold metrics, and a local FastAPI capture demo. + +## Setup + +From the project root: + +```powershell +py -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install -e ".[dev]" +$env:SIGVERIFY_FERNET_KEY = python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +uvicorn app.main:app --reload +``` + +Open . The generated Fernet key is for local testing; +keep the same key between runs to read previously enrolled templates. Store a +production key in a secret manager, never in source control. Without a key, the +API rejects enrollment and verification rather than storing plaintext. + +Run tests with `pytest`. The package can also be installed with `pip install -e .`. + +## Data format + +The common signature format has one sample per point with `user_id`, `t`, `x`, +`y`, optional `p` (pressure), optional `area` (contact area), and optional +`pen_down`. CSV loaders expect these names by default; pass column overrides +for a dataset with different headers. A JSON signature uses the same names +with array-valued `t`, `x`, and `y`. For MOBISIG, check the downloaded release's +layout is supported directly by `load_mobisig`: `timestamp`, `x`, `y`, +`pressure`, and `fingerarea` are mapped automatically. Names such as +`SIGN_GEN_USER1_USER1_1.csv` and `SIGN_FOR_USER1_USER2_1.csv` determine the +claimed user and genuine/forgery label; metadata files such as `users.csv` are +ignored. Put downloaded or self-collected material under `data/raw/`. Raw data +is ignored by Git. + +## Baseline workflow + +1. Load signatures into `Signature`. +2. Sort timestamps, collapse repeated times, lightly smooth, resample, and + translate/scale positions. Preprocessing records the original bounding-box + diagonal because size may itself be discriminative. Pen-up samples are + excluded from the baseline trajectory. +3. Extract x/y and motion features; pressure and contact area can be enabled as + explicit ablations. +4. Enroll several genuine examples, compare using band-constrained DTW, and + divide by the median within-enrollment distance. +5. Tune global and per-user thresholds on a development split; evaluate only + on held-out attempts. Thresholds in the demo are illustrative, not research + results. + +`experiments/` contains starter scripts for inspecting data, running a simple +split, and comparing feature sets. Report user-disjoint development/test +results separately from the API demo. Do not train or tune on the test set. + +For the included MOBISIG layout, run: + +```powershell +python experiments/02_baseline_dtw.py --data-dir data/raw/mobisig ` + --output results/mobisig_baseline.json +``` + +## Capture and consent + +The browser sends pointer positions, elapsed milliseconds, and optional +pressure. It does not persist anything in the browser. Review +[`docs/data_collection_protocol.md`](docs/data_collection_protocol.md) and +[`docs/consent_form.md`](docs/consent_form.md) before collecting data; adapt +them to institutional requirements and obtain approval where required. + +The SQLite demo stores encrypted template JSON and a small lockout counter in +process memory. Lockout state resets when the server restarts, so this is not a +production account-security system. Use HTTPS, authentication, durable +rate-limiting, access controls, key management, and a retention/deletion policy +before exposing a service beyond localhost. diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..987b34a --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""Local FastAPI application.""" diff --git a/app/db.py b/app/db.py new file mode 100644 index 0000000..2782a64 --- /dev/null +++ b/app/db.py @@ -0,0 +1,38 @@ +"""SQLite persistence for encrypted enrollment templates.""" + +import os +from pathlib import Path + +from sqlalchemy import LargeBinary, String, create_engine +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker + +DATABASE_URL = os.environ.get( + "SIGVERIFY_DATABASE_URL", + f"sqlite:///{(Path(__file__).resolve().parents[1] / 'data' / 'signatures.sqlite3').as_posix()}", +) +engine = create_engine( + DATABASE_URL, + connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {}, +) +SessionLocal = sessionmaker(bind=engine, expire_on_commit=False) + + +class Base(DeclarativeBase): + """Declarative base for API database tables.""" + + +class SignatureTemplate(Base): + """An encrypted sample belonging to one claimed user identity.""" + + __tablename__ = "signature_templates" + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + user_id: Mapped[str] = mapped_column(String(200), index=True) + encrypted_payload: Mapped[bytes] = mapped_column(LargeBinary) + + +def initialize_database() -> None: + """Create missing tables and ensure the SQLite data directory exists.""" + if DATABASE_URL.startswith("sqlite:///"): + Path(DATABASE_URL.removeprefix("sqlite:///")).parent.mkdir(parents=True, exist_ok=True) + Base.metadata.create_all(engine) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..77e21c2 --- /dev/null +++ b/app/main.py @@ -0,0 +1,146 @@ +"""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()) + ] + } diff --git a/app/static/capture.js b/app/static/capture.js new file mode 100644 index 0000000..1dce51f --- /dev/null +++ b/app/static/capture.js @@ -0,0 +1,98 @@ +(() => { + const canvas = document.getElementById("pad"); + const context = canvas.getContext("2d"); + const userId = document.getElementById("user-id"); + const status = document.getElementById("status"); + const points = []; + let drawing = false; + let startTime = 0; + + context.lineWidth = 3; + context.lineCap = "round"; + context.lineJoin = "round"; + + function pointFrom(event, penDown) { + const bounds = canvas.getBoundingClientRect(); + const scaleX = canvas.width / bounds.width; + const scaleY = canvas.height / bounds.height; + return { + t: performance.now() - startTime, + x: (event.clientX - bounds.left) * scaleX, + y: (event.clientY - bounds.top) * scaleY, + p: event.pressure || (penDown ? 0.5 : 0), + area: event.width * event.height, + pen_down: penDown, + }; + } + + canvas.addEventListener("pointerdown", (event) => { + event.preventDefault(); + canvas.setPointerCapture(event.pointerId); + if (points.length === 0) startTime = performance.now(); + drawing = true; + const point = pointFrom(event, true); + points.push(point); + context.beginPath(); + context.moveTo(point.x, point.y); + }); + + canvas.addEventListener("pointermove", (event) => { + if (!drawing) return; + const point = pointFrom(event, true); + points.push(point); + context.lineTo(point.x, point.y); + context.stroke(); + }); + + function endStroke(event) { + if (!drawing) return; + points.push(pointFrom(event, false)); + drawing = false; + } + canvas.addEventListener("pointerup", endStroke); + canvas.addEventListener("pointercancel", endStroke); + + document.getElementById("clear").addEventListener("click", () => { + points.length = 0; + context.clearRect(0, 0, canvas.width, canvas.height); + status.textContent = "Signature cleared."; + }); + + async function send(path) { + if (!userId.value.trim()) { + status.textContent = "Enter a participant ID first."; + return; + } + if (points.filter((point) => point.pen_down).length < 2) { + status.textContent = "Draw a signature with at least two points."; + return; + } + const payload = { + user_id: userId.value.trim(), + t: points.map((point) => point.t), + x: points.map((point) => point.x), + y: points.map((point) => point.y), + p: points.map((point) => point.p), + area: points.map((point) => point.area), + pen_down: points.map((point) => point.pen_down), + }; + status.textContent = "Sending…"; + try { + const response = await fetch(path, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }); + const result = await response.json(); + if (!response.ok) throw new Error(result.detail || `Request failed (${response.status})`); + status.textContent = path === "/verify" + ? `Accepted: ${result.accepted}; score: ${result.score.toFixed(3)}; threshold: ${result.threshold.toFixed(3)}` + : `Enrolled sample for ${result.user_id}. Enroll multiple genuine samples.`; + } catch (error) { + status.textContent = `Error: ${error.message}`; + } + } + + document.getElementById("enroll").addEventListener("click", () => send("/enroll")); + document.getElementById("verify").addEventListener("click", () => send("/verify")); +})(); \ No newline at end of file diff --git a/app/static/index.html b/app/static/index.html new file mode 100644 index 0000000..10604a1 --- /dev/null +++ b/app/static/index.html @@ -0,0 +1,27 @@ + + + + + + Signature verification demo + + + +

Signature verification demo

+

Use a study participant ID (not a real name). Draw a signature, then enroll + several genuine samples or verify against the enrolled templates.

+ + + + + + +

+ + + \ No newline at end of file diff --git a/data/processed/.gitkeep b/data/processed/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/raw/mobisig/.gitkeep b/data/raw/mobisig/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/raw/mobisig/statistics.txt b/data/raw/mobisig/statistics.txt new file mode 100644 index 0000000..7fa1337 --- /dev/null +++ b/data/raw/mobisig/statistics.txt @@ -0,0 +1,8 @@ +Statistics + +All: 83 +Male: 49 +Female: 34 + +Age: 26.03 (9.84) + diff --git a/data/raw/own/.gitkeep b/data/raw/own/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/docs/consent_form.md b/docs/consent_form.md new file mode 100644 index 0000000..6cf8650 --- /dev/null +++ b/docs/consent_form.md @@ -0,0 +1,45 @@ +# Participant information and consent (draft) + +**This is a study-material template, not legal advice or institutional approval.** +Adapt it to the actual study, data controller, storage location, retention +period, withdrawal process, and applicable ethics/privacy requirements before +use. Do not collect signatures until required approvals have been obtained. + +## Study + +Study title: Online handwritten-signature verification +Researcher/contact: ____________________________________ +Institution/contact: ___________________________________ + +## What participation involves + +You may be asked to provide several genuine signatures and, if applicable, +clearly identified simulated forgeries. The capture may include pen/cursor +position, timestamps, pressure, and contact area when the device provides them. +The purpose is to evaluate a research algorithm; the system is experimental and +must not be used for legal identity decisions. + +## Data use and protection + +The study team will use the collected data for the stated research purpose. +Signature dynamics can be sensitive biometric information. The project should +use participant codes instead of names, limit access, encrypt stored templates, +set a deletion date, and explain any sharing or publication of derived results. +The researcher must fill in the actual retention and deletion arrangements +before presenting this form. + +## Voluntary participation + +Participation is voluntary. You may stop at any time and may request removal of +your identifiable raw records until: ____________________. Explain any limits +on withdrawal after data have been anonymized or published. + +## Consent + +I have read and understood the information above, had the opportunity to ask +questions, and voluntarily agree to participate. + +Participant code: ____________________ +Participant signature: ____________________ Date: __________ +Researcher signature: ____________________ Date: __________ + diff --git a/docs/data_collection_protocol.md b/docs/data_collection_protocol.md new file mode 100644 index 0000000..9045a54 --- /dev/null +++ b/docs/data_collection_protocol.md @@ -0,0 +1,25 @@ +# Data collection protocol (draft) + +1. Obtain institutional ethics/privacy review where required and use the + completed participant information and consent form. +2. Assign a random participant code. Do not put names or contact details in + filenames or the signature database. +3. Record device type, sampling setup, session date/code, and whether pressure + or contact-area data are available in a separate access-controlled study + log. Do not write unnecessary personal information into the signature CSV. +4. Collect several genuine samples per participant, with a short pause between + attempts. Mark examples explicitly as `genuine`. +5. If studying skilled or random forgeries, collect them only under the approved + protocol, keep the target identity in `user_id`, and label their type in the + `label` column. Avoid coercive or deceptive collection. +6. Use the common CSV columns `user_id,label,t,x,y,p,area,pen_down`. Each file + should contain one signature sequence; timestamp units must be consistent + within a dataset. Optional `p` and `area` fields should be blank or omitted + when unsupported. +7. Store raw files in `data/raw/own/`, restrict access, and keep them out of + version control. Store encryption keys separately from the database. +8. Keep participant-level identities together when making train/test splits. + Tune thresholds on a development set and report the final test set only + once. Report the number of participants, genuine samples, and each forgery + type alongside FAR, FRR, and EER. + diff --git a/experiments/01_explore_data.ipynb b/experiments/01_explore_data.ipynb new file mode 100644 index 0000000..17c9198 --- /dev/null +++ b/experiments/01_explore_data.ipynb @@ -0,0 +1,55 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Explore signature captures\n", + "\n", + "Set `DATA_DIR` to a folder of common-format CSV signatures. Raw dataset formats may need a column mapping in `sigverify.io.loaders` first." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from pathlib import Path\n", + "import matplotlib.pyplot as plt\n", + "from sigverify.io.loaders import load_csv_directory\n", + "\n", + "DATA_DIR = Path('../data/raw/mobisig')\n", + "signatures = load_csv_directory(DATA_DIR)\n", + "len(signatures)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "for signature in signatures[:6]:\n", + " plt.plot(signature.x, signature.y, label=f'{signature.user_id} ({signature.label})')\n", + "plt.gca().invert_yaxis()\n", + "plt.axis('equal')\n", + "plt.legend()\n", + "plt.show()" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/experiments/02_baseline_dtw.py b/experiments/02_baseline_dtw.py new file mode 100644 index 0000000..e39c8af --- /dev/null +++ b/experiments/02_baseline_dtw.py @@ -0,0 +1,44 @@ +"""Run the DTW baseline on a folder of per-signature CSV files.""" + +import argparse +import json +from pathlib import Path + +try: + from experiments.common import evaluate_directory +except ModuleNotFoundError: + from common import evaluate_directory + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-dir", default="data/raw/mobisig") + parser.add_argument("--enrollment-count", type=int, default=3) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--output", type=Path) + args = parser.parse_args() + metrics = evaluate_directory( + args.data_dir, enrollment_count=args.enrollment_count, seed=args.seed, mobisig=True + ) + print(f"EER: {metrics['eer']:.4f}") + print(f"Threshold at approximate EER: {metrics['eer_threshold']:.4f}") + if args.output: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text( + json.dumps( + { + "dataset": str(args.data_dir), + "enrollment_count": args.enrollment_count, + "seed": args.seed, + "eer": metrics["eer"], + "eer_threshold": metrics["eer_threshold"], + }, + indent=2, + ) + + "\n", + encoding="utf-8", + ) + + +if __name__ == "__main__": + main() diff --git a/experiments/03_feature_ablation.py b/experiments/03_feature_ablation.py new file mode 100644 index 0000000..05e7528 --- /dev/null +++ b/experiments/03_feature_ablation.py @@ -0,0 +1,34 @@ +"""Compare DTW with optional pressure and contact-area channels.""" + +import argparse + +try: + from experiments.common import evaluate_directory +except ModuleNotFoundError: + from common import evaluate_directory + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-dir", default="data/raw/mobisig") + parser.add_argument("--enrollment-count", type=int, default=3) + args = parser.parse_args() + configurations = [ + ("position and motion", False, False), + ("plus pressure", True, False), + ("plus contact area", False, True), + ("pressure and area", True, True), + ] + for name, pressure, area in configurations: + metrics = evaluate_directory( + args.data_dir, + enrollment_count=args.enrollment_count, + include_pressure=pressure, + include_area=area, + mobisig=True, + ) + print(f"{name}: EER={metrics['eer']:.4f}") + + +if __name__ == "__main__": + main() diff --git a/experiments/04_own_data_eval.py b/experiments/04_own_data_eval.py new file mode 100644 index 0000000..40edb59 --- /dev/null +++ b/experiments/04_own_data_eval.py @@ -0,0 +1,25 @@ +"""Evaluate the same DTW baseline on locally collected signatures.""" + +import argparse + +try: + from experiments.common import evaluate_directory +except ModuleNotFoundError: + from common import evaluate_directory + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--data-dir", default="data/raw/own") + parser.add_argument("--enrollment-count", type=int, default=3) + parser.add_argument("--seed", type=int, default=0) + args = parser.parse_args() + metrics = evaluate_directory( + args.data_dir, enrollment_count=args.enrollment_count, seed=args.seed + ) + print(f"Own-data EER: {metrics['eer']:.4f}") + print(f"Threshold at approximate EER: {metrics['eer_threshold']:.4f}") + + +if __name__ == "__main__": + main() diff --git a/experiments/__init__.py b/experiments/__init__.py new file mode 100644 index 0000000..47ec177 --- /dev/null +++ b/experiments/__init__.py @@ -0,0 +1,2 @@ +"""Reproducible baseline experiment entry points.""" + diff --git a/experiments/common.py b/experiments/common.py new file mode 100644 index 0000000..d47b7bf --- /dev/null +++ b/experiments/common.py @@ -0,0 +1,45 @@ +"""Shared helpers for command-line signature evaluation scripts.""" + +from collections import defaultdict +from pathlib import Path + +from sigverify.evaluation.metrics import verification_metrics +from sigverify.evaluation.protocols import enrollment_test_split +from sigverify.io.loaders import load_csv_directory, load_mobisig +from sigverify.io.schema import Signature +from sigverify.matching.models import DTWVerifier +from sigverify.preprocessing import preprocess_signature + + +def evaluate_directory( + directory: str | Path, + *, + enrollment_count: int = 3, + include_pressure: bool = False, + include_area: bool = False, + seed: int = 0, + mobisig: bool = False, +) -> dict[str, object]: + """Load, preprocess, split, and score known-identity attempts.""" + raw = load_mobisig(directory) if mobisig else load_csv_directory(directory) + samples = [ + preprocess_signature(signature) + for signature in raw + ] + enrollment, test = enrollment_test_split( + samples, enrollment_count=enrollment_count, seed=seed + ) + templates: dict[str, list[Signature]] = defaultdict(list) + for signature in enrollment: + templates[signature.user_id].append(signature) + + labels: list[int] = [] + scores: list[float] = [] + for sample in test: + if sample.user_id not in templates: + raise ValueError(f"test identity {sample.user_id!r} has no enrollment samples") + verifier = DTWVerifier(include_pressure=include_pressure, include_area=include_area) + verifier.enroll(templates[sample.user_id]) + scores.append(verifier.score(sample)) + labels.append(int(sample.label.lower() in {"genuine", "g", "1"})) + return verification_metrics(labels, scores) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..de9d110 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "sigverify" +version = "0.1.0" +description = "A small online handwritten-signature verification baseline" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "cryptography>=42", + "fastapi>=0.110", + "matplotlib>=3.8", + "numpy>=1.26", + "pandas>=2.1", + "pydantic>=2.6", + "scikit-learn>=1.4", + "scipy>=1.12", + "sqlalchemy>=2.0", + "uvicorn[standard]>=0.27", +] + +[project.optional-dependencies] +dev = ["httpx>=0.27", "pytest>=8"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] + diff --git a/results/.gitkeep b/results/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/src/sigverify.egg-info/PKG-INFO b/src/sigverify.egg-info/PKG-INFO new file mode 100644 index 0000000..4971f03 --- /dev/null +++ b/src/sigverify.egg-info/PKG-INFO @@ -0,0 +1,89 @@ +Metadata-Version: 2.4 +Name: sigverify +Version: 0.1.0 +Summary: A small online handwritten-signature verification baseline +Requires-Python: >=3.10 +Description-Content-Type: text/markdown +Requires-Dist: cryptography>=42 +Requires-Dist: fastapi>=0.110 +Requires-Dist: matplotlib>=3.8 +Requires-Dist: numpy>=1.26 +Requires-Dist: pandas>=2.1 +Requires-Dist: pydantic>=2.6 +Requires-Dist: scikit-learn>=1.4 +Requires-Dist: scipy>=1.12 +Requires-Dist: sqlalchemy>=2.0 +Requires-Dist: uvicorn[standard]>=0.27 +Provides-Extra: dev +Requires-Dist: httpx>=0.27; extra == "dev" +Requires-Dist: pytest>=8; extra == "dev" + +# Signature verification baseline + +A small research starter project for online handwritten-signature verification. +It includes common-format CSV/JSON loading, preprocessing, feature extraction, +DTW matching, threshold metrics, and a local FastAPI capture demo. + +## Setup + +From the project root: + +```powershell +py -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install -e ".[dev]" +$env:SIGVERIFY_FERNET_KEY = python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +uvicorn app.main:app --reload +``` + +Open . The generated Fernet key is for local testing; +keep the same key between runs to read previously enrolled templates. Store a +production key in a secret manager, never in source control. Without a key, the +API rejects enrollment and verification rather than storing plaintext. + +Run tests with `pytest`. The package can also be installed with `pip install -e .`. + +## Data format + +The common signature format has one sample per point with `user_id`, `t`, `x`, +`y`, optional `p` (pressure), optional `area` (contact area), and optional +`pen_down`. CSV loaders expect these names by default; pass column overrides +for a dataset with different headers. A JSON signature uses the same names +with array-valued `t`, `x`, and `y`. For MOBISIG, check the downloaded release's +file layout and map its columns with the loader arguments; dataset releases can +vary. Put downloaded or self-collected material under `data/raw/`. Raw data is +ignored by Git. + +## Baseline workflow + +1. Load signatures into `Signature`. +2. Sort timestamps, collapse repeated times, lightly smooth, resample, and + translate/scale positions. Preprocessing records the original bounding-box + diagonal because size may itself be discriminative. Pen-up samples are + excluded from the baseline trajectory. +3. Extract x/y and motion features; pressure and contact area can be enabled as + explicit ablations. +4. Enroll several genuine examples, compare using band-constrained DTW, and + divide by the median within-enrollment distance. +5. Tune global and per-user thresholds on a development split; evaluate only + on held-out attempts. Thresholds in the demo are illustrative, not research + results. + +`experiments/` contains starter scripts for inspecting data, running a simple +split, and comparing feature sets. Report user-disjoint development/test +results separately from the API demo. Do not train or tune on the test set. + +## Capture and consent + +The browser sends pointer positions, elapsed milliseconds, and optional +pressure. It does not persist anything in the browser. Review +[`docs/data_collection_protocol.md`](docs/data_collection_protocol.md) and +[`docs/consent_form.md`](docs/consent_form.md) before collecting data; adapt +them to institutional requirements and obtain approval where required. + +The SQLite demo stores encrypted template JSON and a small lockout counter in +process memory. Lockout state resets when the server restarts, so this is not a +production account-security system. Use HTTPS, authentication, durable +rate-limiting, access controls, key management, and a retention/deletion policy +before exposing a service beyond localhost. + diff --git a/src/sigverify.egg-info/SOURCES.txt b/src/sigverify.egg-info/SOURCES.txt new file mode 100644 index 0000000..e646a92 --- /dev/null +++ b/src/sigverify.egg-info/SOURCES.txt @@ -0,0 +1,21 @@ +README.md +pyproject.toml +src/sigverify/__init__.py +src/sigverify/decision.py +src/sigverify/features.py +src/sigverify/preprocessing.py +src/sigverify/security.py +src/sigverify.egg-info/PKG-INFO +src/sigverify.egg-info/SOURCES.txt +src/sigverify.egg-info/dependency_links.txt +src/sigverify.egg-info/requires.txt +src/sigverify.egg-info/top_level.txt +src/sigverify/evaluation/metrics.py +src/sigverify/evaluation/plots.py +src/sigverify/evaluation/protocols.py +src/sigverify/io/__init__.py +src/sigverify/io/loaders.py +src/sigverify/io/schema.py +src/sigverify/matching/dtw.py +src/sigverify/matching/models.py +tests/test_baseline.py \ No newline at end of file diff --git a/src/sigverify.egg-info/dependency_links.txt b/src/sigverify.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/sigverify.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/src/sigverify.egg-info/requires.txt b/src/sigverify.egg-info/requires.txt new file mode 100644 index 0000000..b6fde48 --- /dev/null +++ b/src/sigverify.egg-info/requires.txt @@ -0,0 +1,14 @@ +cryptography>=42 +fastapi>=0.110 +matplotlib>=3.8 +numpy>=1.26 +pandas>=2.1 +pydantic>=2.6 +scikit-learn>=1.4 +scipy>=1.12 +sqlalchemy>=2.0 +uvicorn[standard]>=0.27 + +[dev] +httpx>=0.27 +pytest>=8 diff --git a/src/sigverify.egg-info/top_level.txt b/src/sigverify.egg-info/top_level.txt new file mode 100644 index 0000000..50381a1 --- /dev/null +++ b/src/sigverify.egg-info/top_level.txt @@ -0,0 +1 @@ +sigverify diff --git a/src/sigverify/__init__.py b/src/sigverify/__init__.py new file mode 100644 index 0000000..ccd5c5a --- /dev/null +++ b/src/sigverify/__init__.py @@ -0,0 +1,2 @@ +"""Online handwritten-signature verification research starter.""" + diff --git a/src/sigverify/decision.py b/src/sigverify/decision.py new file mode 100644 index 0000000..48a7642 --- /dev/null +++ b/src/sigverify/decision.py @@ -0,0 +1,19 @@ +"""Global and per-user threshold decisions.""" + +from dataclasses import dataclass, field + + +@dataclass(slots=True) +class Thresholds: + """Threshold registry; scores at or below a threshold are accepted.""" + + global_threshold: float + user_thresholds: dict[str, float] = field(default_factory=dict) + + def threshold_for(self, user_id: str, *, user_dependent: bool = True) -> float: + if user_dependent and user_id in self.user_thresholds: + return self.user_thresholds[user_id] + return self.global_threshold + + def accept(self, user_id: str, score: float, *, user_dependent: bool = True) -> bool: + return score <= self.threshold_for(user_id, user_dependent=user_dependent) diff --git a/src/sigverify/evaluation/metrics.py b/src/sigverify/evaluation/metrics.py new file mode 100644 index 0000000..de05e93 --- /dev/null +++ b/src/sigverify/evaluation/metrics.py @@ -0,0 +1,36 @@ +"""Threshold-independent and threshold-dependent verification metrics.""" + +import numpy as np +from numpy.typing import ArrayLike +from sklearn.metrics import det_curve, roc_curve + + +def verification_metrics(labels: ArrayLike, scores: ArrayLike) -> dict[str, object]: + """Compute FAR/FRR curves, ROC, DET and the approximate EER. + + Labels use 0 for forgery and 1 for genuine. Scores are lower for more + genuine-looking samples. + """ + y = np.asarray(labels, dtype=np.int8) + values = np.asarray(scores, dtype=np.float64) + if y.ndim != 1 or values.ndim != 1 or len(y) != len(values) or len(y) == 0: + raise ValueError("labels and scores must be non-empty 1D arrays of equal length") + if not np.isin(y, [0, 1]).all() or len(np.unique(y)) != 2: + raise ValueError("labels must contain both classes, encoded as 0=forgery and 1=genuine") + + fpr, tpr, roc_thresholds = roc_curve(y, -values) + false_accept, false_reject, det_thresholds = det_curve(y, -values) + fnr = 1.0 - tpr + crossing = int(np.argmin(np.abs(fpr - fnr))) + return { + "far": fpr, + "frr": fnr, + "roc_fpr": fpr, + "roc_tpr": tpr, + "roc_thresholds": -roc_thresholds, + "det_fpr": false_accept, + "det_fnr": false_reject, + "det_thresholds": -det_thresholds, + "eer": float((fpr[crossing] + fnr[crossing]) / 2.0), + "eer_threshold": float(-roc_thresholds[crossing]), + } diff --git a/src/sigverify/evaluation/plots.py b/src/sigverify/evaluation/plots.py new file mode 100644 index 0000000..9235cfd --- /dev/null +++ b/src/sigverify/evaluation/plots.py @@ -0,0 +1,33 @@ +"""Plot ROC and DET curves for evaluation results.""" + +from pathlib import Path + +import matplotlib.pyplot as plt + + +def plot_roc(metrics: dict[str, object], destination: str | Path) -> None: + """Write a ROC curve image from `verification_metrics` output.""" + target = Path(destination) + target.parent.mkdir(parents=True, exist_ok=True) + plt.figure() + plt.plot(metrics["roc_fpr"], metrics["roc_tpr"], label=f"EER={metrics['eer']:.3f}") + plt.plot([0, 1], [0, 1], "--", color="gray") + plt.xlabel("False acceptance rate") + plt.ylabel("True acceptance rate") + plt.legend() + plt.tight_layout() + plt.savefig(target) + plt.close() + + +def plot_det(metrics: dict[str, object], destination: str | Path) -> None: + """Write a DET curve image from `verification_metrics` output.""" + target = Path(destination) + target.parent.mkdir(parents=True, exist_ok=True) + plt.figure() + plt.plot(metrics["det_fpr"], metrics["det_fnr"]) + plt.xlabel("False acceptance rate") + plt.ylabel("False rejection rate") + plt.tight_layout() + plt.savefig(target) + plt.close() diff --git a/src/sigverify/evaluation/protocols.py b/src/sigverify/evaluation/protocols.py new file mode 100644 index 0000000..b979c2b --- /dev/null +++ b/src/sigverify/evaluation/protocols.py @@ -0,0 +1,51 @@ +"""Simple enrollment/test and user-disjoint evaluation splits.""" + +from collections import defaultdict +from collections.abc import Sequence + +import numpy as np + +from sigverify.io.schema import Signature + + +def enrollment_test_split( + samples: Sequence[Signature], + *, + enrollment_count: int = 3, + seed: int = 0, +) -> tuple[list[Signature], list[Signature]]: + """Split genuine samples per user; non-genuine samples remain in test.""" + if enrollment_count < 1: + raise ValueError("enrollment_count must be positive") + groups: dict[str, list[Signature]] = defaultdict(list) + test = [sample for sample in samples if sample.label.lower() not in {"genuine", "g", "1"}] + for sample in samples: + if sample.label.lower() in {"genuine", "g", "1"}: + groups[sample.user_id].append(sample) + rng = np.random.default_rng(seed) + enrollment: list[Signature] = [] + for user_id, group in groups.items(): + if len(group) <= enrollment_count: + raise ValueError(f"user {user_id!r} needs more than {enrollment_count} genuine samples") + order = rng.permutation(len(group)) + enrollment.extend(group[index] for index in order[:enrollment_count]) + test.extend(group[index] for index in order[enrollment_count:]) + return enrollment, test + + +def user_disjoint_split( + samples: Sequence[Signature], *, test_fraction: float = 0.2, seed: int = 0 +) -> tuple[list[Signature], list[Signature]]: + """Split complete user identities, suitable for learning-model evaluation.""" + if not 0 < test_fraction < 1: + raise ValueError("test_fraction must be between 0 and 1") + users = np.asarray(sorted({sample.user_id for sample in samples})) + if len(users) < 2: + raise ValueError("at least two users are required for a user-disjoint split") + test_users = set(np.random.default_rng(seed).permutation(users)[: max( + 1, min(len(users) - 1, round(len(users) * test_fraction)) + )]) + return ( + [sample for sample in samples if sample.user_id not in test_users], + [sample for sample in samples if sample.user_id in test_users], + ) diff --git a/src/sigverify/features.py b/src/sigverify/features.py new file mode 100644 index 0000000..48f22cf --- /dev/null +++ b/src/sigverify/features.py @@ -0,0 +1,56 @@ +"""Per-point trajectory features and per-signature global descriptors.""" + +import numpy as np +from numpy.typing import NDArray + +from sigverify.io.schema import Signature + + +def extract_features( + signature: Signature, + *, + include_pressure: bool = False, + include_area: bool = False, + include_acceleration: bool = True, + include_curvature: bool = True, +) -> NDArray[np.float64]: + """Return point features; x/y, velocity and speed are the initial channels.""" + elapsed = signature.t - signature.t[0] + dt = np.maximum(np.gradient(elapsed), 1e-6) + + def derivative(values: NDArray[np.float64]) -> NDArray[np.float64]: + if len(values) < 2: + return np.zeros_like(values) + return np.gradient(values) / dt + + dx = derivative(signature.x) + dy = derivative(signature.y) + columns = [signature.x, signature.y, dx, dy, np.hypot(dx, dy)] + if include_acceleration: + columns.append(derivative(columns[-1])) + if include_curvature: + ddx = derivative(dx) + ddy = derivative(dy) + denominator = np.maximum((dx * dx + dy * dy) ** 1.5, 1e-8) + columns.append((dx * ddy - dy * ddx) / denominator) + if include_pressure: + if signature.p is None: + raise ValueError("pressure requested but this signature has no pressure data") + columns.append(signature.p) + if include_area: + if signature.area is None: + raise ValueError("area requested but this signature has no area data") + columns.append(signature.area) + return np.column_stack(columns).astype(np.float64, copy=False) + + +def global_features(signature: Signature) -> dict[str, float]: + """Return duration, path length, and original size where available.""" + path_length = float(np.hypot(np.diff(signature.x), np.diff(signature.y)).sum()) + result = { + "duration": float(signature.t[-1] - signature.t[0]), + "path_length": path_length, + } + if signature.original_size is not None: + result["original_size"] = signature.original_size + return result diff --git a/src/sigverify/io/__init__.py b/src/sigverify/io/__init__.py new file mode 100644 index 0000000..3af79a4 --- /dev/null +++ b/src/sigverify/io/__init__.py @@ -0,0 +1,2 @@ +"""Data schemas and loaders.""" + diff --git a/src/sigverify/io/loaders.py b/src/sigverify/io/loaders.py new file mode 100644 index 0000000..614017c --- /dev/null +++ b/src/sigverify/io/loaders.py @@ -0,0 +1,136 @@ +"""Load CSV and JSON captures into the common signature schema.""" + +import json +from pathlib import Path + +import pandas as pd + +from sigverify.io.schema import Signature + + +def load_csv_directory( + directory: str | Path, + *, + user_column: str = "user_id", + time_column: str = "t", + x_column: str = "x", + y_column: str = "y", + pressure_column: str | None = "p", + area_column: str | None = "area", + pen_down_column: str | None = "pen_down", + label_column: str | None = "label", +) -> list[Signature]: + """Load CSV files; each file may contain one or multiple users.""" + root = Path(directory) + if not root.is_dir(): + raise FileNotFoundError(f"signature directory does not exist: {root}") + + signatures: list[Signature] = [] + for path in sorted(root.rglob("*.csv")): + frame = pd.read_csv(path) + required = [time_column, x_column, y_column] + missing = [column for column in required if column not in frame.columns] + if missing: + raise ValueError(f"{path} is missing required columns: {', '.join(missing)}") + if user_column not in frame.columns: + frame[user_column] = path.stem + + for user_id, group in frame.groupby(user_column, sort=False, dropna=False): + def optional_column(column: str | None): + return group[column].to_numpy() if column and column in group.columns else None + + labels = group[label_column].dropna().unique() if label_column and label_column in group else [] + label = str(labels[0]) if len(labels) else "genuine" + signatures.append( + Signature( + user_id=str(user_id), + label=label, + t=group[time_column].to_numpy(), + x=group[x_column].to_numpy(), + y=group[y_column].to_numpy(), + p=optional_column(pressure_column), + area=optional_column(area_column), + pen_down=optional_column(pen_down_column), + ) + ) + return signatures + + +def load_json_signature(path: str | Path) -> Signature: + """Load one signature from a JSON object in the common format.""" + source = Path(path) + with source.open(encoding="utf-8") as stream: + data = json.load(stream) + return Signature( + user_id=str(data["user_id"]), + label=str(data.get("label", "genuine")), + t=data["t"], + x=data["x"], + y=data["y"], + p=data.get("p"), + area=data.get("area"), + pen_down=data.get("pen_down"), + ) + + +def load_mobisig(directory: str | Path, **column_mapping: str | None) -> list[Signature]: + """Load the released MOBISIG directory layout. + + MOBISIG files encode the claimed signer and genuine/forgery label in names + such as ``SIGN_GEN_USER1_USER1_1.csv`` and + ``SIGN_FOR_USER1_USER2_1.csv``. The parent directory is used as a + fallback identity, while the filename's claimed signer is authoritative. + """ + root = Path(directory) + if not root.is_dir(): + raise FileNotFoundError(f"signature directory does not exist: {root}") + mapping = { + "time": "timestamp", + "x": "x", + "y": "y", + "pressure": "pressure", + "area": "fingerarea", + } + mapping.update({key: value for key, value in column_mapping.items() if value is not None}) + signatures: list[Signature] = [] + for path in sorted(root.rglob("*.csv")): + if not path.stem.upper().startswith("SIGN_"): + continue + frame = pd.read_csv(path) + required = [mapping["time"], mapping["x"], mapping["y"]] + missing = [column for column in required if column not in frame.columns] + if missing: + raise ValueError(f"{path} is missing MOBISIG columns: {', '.join(missing)}") + + parts = path.stem.upper().split("_") + if len(parts) >= 4 and parts[1] in {"GEN", "FOR"}: + label = "genuine" if parts[1] == "GEN" else "forgery" + user_id = parts[2] + else: + label = "genuine" + user_id = path.parent.name + signatures.append( + Signature( + user_id=user_id, + label=label, + t=frame[mapping["time"]].to_numpy(dtype=float), + x=frame[mapping["x"]].to_numpy(dtype=float), + y=frame[mapping["y"]].to_numpy(dtype=float), + p=( + frame[mapping["pressure"]].to_numpy(dtype=float) + if mapping["pressure"] in frame + else None + ), + area=( + frame[mapping["area"]].to_numpy(dtype=float) + if mapping["area"] in frame + else None + ), + ) + ) + return signatures + + +def load_own_data(directory: str | Path) -> list[Signature]: + """Load locally captured CSVs using the common column names.""" + return load_csv_directory(directory) diff --git a/src/sigverify/io/schema.py b/src/sigverify/io/schema.py new file mode 100644 index 0000000..f8fa045 --- /dev/null +++ b/src/sigverify/io/schema.py @@ -0,0 +1,71 @@ +"""Common representation for time-series signature captures.""" + +from dataclasses import dataclass + +import numpy as np +from numpy.typing import NDArray + +FloatArray = NDArray[np.float64] +BoolArray = NDArray[np.bool_] + + +@dataclass(slots=True) +class Signature: + """One captured signature, represented as equally sized point arrays.""" + + user_id: str + label: str + t: FloatArray + x: FloatArray + y: FloatArray + p: FloatArray | None = None + area: FloatArray | None = None + pen_down: BoolArray | None = None + original_size: float | None = None + size_normalized: bool = False + + def __post_init__(self) -> None: + self.t = np.asarray(self.t, dtype=np.float64) + self.x = np.asarray(self.x, dtype=np.float64) + self.y = np.asarray(self.y, dtype=np.float64) + for name in ("p", "area"): + value = getattr(self, name) + if value is not None: + setattr(self, name, np.asarray(value, dtype=np.float64)) + if self.pen_down is not None: + self.pen_down = np.asarray(self.pen_down, dtype=np.bool_) + + n = self.t.size + if not self.user_id: + raise ValueError("user_id must not be empty") + if n == 0: + raise ValueError("a signature must contain at least one point") + for name in ("x", "y", "p", "area", "pen_down"): + value = getattr(self, name) + if value is not None: + if value.ndim != 1: + raise ValueError(f"{name} must be a one-dimensional array") + if value.size != n: + raise ValueError(f"{name} must have the same number of points as t") + if not (np.isfinite(self.t).all() and np.isfinite(self.x).all() and np.isfinite(self.y).all()): + raise ValueError("t, x, and y must contain only finite values") + for name in ("p", "area"): + value = getattr(self, name) + if value is not None and not np.isfinite(value).all(): + raise ValueError(f"{name} must contain only finite values") + + def copy(self, **changes: object) -> "Signature": + values = { + "user_id": self.user_id, + "label": self.label, + "t": self.t.copy(), + "x": self.x.copy(), + "y": self.y.copy(), + "p": None if self.p is None else self.p.copy(), + "area": None if self.area is None else self.area.copy(), + "pen_down": None if self.pen_down is None else self.pen_down.copy(), + "original_size": self.original_size, + "size_normalized": self.size_normalized, + } + values.update(changes) + return Signature(**values) diff --git a/src/sigverify/matching/dtw.py b/src/sigverify/matching/dtw.py new file mode 100644 index 0000000..fd806e7 --- /dev/null +++ b/src/sigverify/matching/dtw.py @@ -0,0 +1,35 @@ +"""Dynamic time warping with an optional Sakoe-Chiba constraint.""" + +import numpy as np +from numpy.typing import ArrayLike + + +def dtw_distance( + first: ArrayLike, + second: ArrayLike, + *, + band_fraction: float | None = 0.15, +) -> float: + """Compute Euclidean local-cost DTW, normalized by the longer sequence.""" + a = np.asarray(first, dtype=np.float64) + b = np.asarray(second, dtype=np.float64) + if a.ndim == 1: + a = a[:, None] + if b.ndim == 1: + b = b[:, None] + if a.ndim != 2 or b.ndim != 2 or a.shape[1] != b.shape[1]: + raise ValueError("sequences must be 1D or 2D with matching feature counts") + if not len(a) or not len(b): + raise ValueError("sequences must not be empty") + if band_fraction is not None and band_fraction < 0: + raise ValueError("band_fraction must be non-negative or None") + + n, m = len(a), len(b) + band = max(abs(n - m), int(max(n, m) * band_fraction)) if band_fraction is not None else max(n, m) + costs = np.full((n + 1, m + 1), np.inf, dtype=np.float64) + costs[0, 0] = 0.0 + for i in range(1, n + 1): + for j in range(max(1, i - band), min(m, i + band) + 1): + local = float(np.linalg.norm(a[i - 1] - b[j - 1])) + costs[i, j] = local + min(costs[i - 1, j], costs[i, j - 1], costs[i - 1, j - 1]) + return float(costs[n, m] / max(n, m)) diff --git a/src/sigverify/matching/models.py b/src/sigverify/matching/models.py new file mode 100644 index 0000000..216ad9f --- /dev/null +++ b/src/sigverify/matching/models.py @@ -0,0 +1,75 @@ +"""Verifier interface and a DTW nearest-enrollment baseline.""" + +from abc import ABC, abstractmethod +from collections.abc import Sequence + +import numpy as np +from numpy.typing import NDArray + +from sigverify.features import extract_features +from sigverify.io.schema import Signature +from sigverify.matching.dtw import dtw_distance + + +class Verifier(ABC): + """Minimal interface implemented by signature verifiers.""" + + @abstractmethod + def enroll(self, samples: Sequence[Signature]) -> None: + """Fit an identity template from genuine enrollment samples.""" + + @abstractmethod + def score(self, sample: Signature) -> float: + """Return a lower-is-more-genuine comparison score.""" + + +class DTWVerifier(Verifier): + """Nearest-template DTW score normalized by enrollment variability.""" + + def __init__( + self, + *, + band_fraction: float = 0.15, + aggregation: str = "minimum", + include_pressure: bool = False, + include_area: bool = False, + ) -> None: + if aggregation not in {"minimum", "mean"}: + raise ValueError("aggregation must be 'minimum' or 'mean'") + self.band_fraction = band_fraction + self.aggregation = aggregation + self.include_pressure = include_pressure + self.include_area = include_area + self._templates: list[NDArray[np.float64]] = [] + self._within_distance = 1.0 + + def _features(self, signature: Signature) -> NDArray[np.float64]: + return extract_features( + signature, + include_pressure=self.include_pressure, + include_area=self.include_area, + ) + + def enroll(self, samples: Sequence[Signature]) -> None: + if not samples: + raise ValueError("at least one enrollment sample is required") + if any(sample.label.lower() not in {"genuine", "g", "1"} for sample in samples): + raise ValueError("enrollment samples must be labelled genuine") + self._templates = [self._features(sample) for sample in samples] + distances = [ + dtw_distance(a, b, band_fraction=self.band_fraction) + for index, a in enumerate(self._templates) + for b in self._templates[index + 1 :] + ] + self._within_distance = max(float(np.median(distances)) if distances else 1.0, 1e-8) + + def score(self, sample: Signature) -> float: + if not self._templates: + raise RuntimeError("enroll must be called before score") + test = self._features(sample) + distances = [ + dtw_distance(template, test, band_fraction=self.band_fraction) + for template in self._templates + ] + aggregate = min(distances) if self.aggregation == "minimum" else float(np.mean(distances)) + return float(aggregate / self._within_distance) diff --git a/src/sigverify/preprocessing.py b/src/sigverify/preprocessing.py new file mode 100644 index 0000000..9edab24 --- /dev/null +++ b/src/sigverify/preprocessing.py @@ -0,0 +1,76 @@ +"""Cleaning, smoothing, resampling, and spatial normalization.""" + +import numpy as np +from scipy.ndimage import gaussian_filter1d + +from sigverify.io.schema import Signature + + +def preprocess_signature( + signature: Signature, + *, + points: int = 100, + smoothing_sigma: float = 0.8, + normalize_position: bool = True, +) -> Signature: + """Return a cleaned signature; original size is retained as metadata.""" + if points < 2: + raise ValueError("points must be at least 2") + if smoothing_sigma < 0: + raise ValueError("smoothing_sigma must not be negative") + + order = np.argsort(signature.t, kind="stable") + sorted_t = signature.t[order] + order = order[np.r_[True, np.diff(sorted_t) > 0]] + pen_down = None if signature.pen_down is None else signature.pen_down[order] + if pen_down is not None: + order = order[pen_down] + if order.size == 0: + raise ValueError("signature has no pen-down points") + + t = signature.t[order] + x = signature.x[order].copy() + y = signature.y[order].copy() + p = None if signature.p is None else signature.p[order].copy() + area = None if signature.area is None else signature.area[order].copy() + original_size = float(np.hypot(np.ptp(x), np.ptp(y))) + + if smoothing_sigma and t.size > 2: + x = gaussian_filter1d(x, smoothing_sigma, mode="nearest") + y = gaussian_filter1d(y, smoothing_sigma, mode="nearest") + if p is not None: + p = gaussian_filter1d(p, smoothing_sigma, mode="nearest") + + if t.size == 1: + new_t = np.linspace(t[0], t[0] + 1.0, points) + x = np.repeat(x, points) + y = np.repeat(y, points) + p = None if p is None else np.repeat(p, points) + area = None if area is None else np.repeat(area, points) + else: + new_t = np.linspace(t[0], t[-1], points) + x = np.interp(new_t, t, x) + y = np.interp(new_t, t, y) + p = None if p is None else np.interp(new_t, t, p) + area = None if area is None else np.interp(new_t, t, area) + + if normalize_position: + x -= float(np.mean(x)) + y -= float(np.mean(y)) + scale = max(float(np.ptp(x)), float(np.ptp(y))) + if scale > np.finfo(np.float64).eps: + x /= scale + y /= scale + + return Signature( + user_id=signature.user_id, + label=signature.label, + t=new_t, + x=x, + y=y, + p=p, + area=area, + pen_down=np.ones(points, dtype=np.bool_), + original_size=original_size, + size_normalized=normalize_position, + ) diff --git a/src/sigverify/security.py b/src/sigverify/security.py new file mode 100644 index 0000000..5f1e67f --- /dev/null +++ b/src/sigverify/security.py @@ -0,0 +1,72 @@ +"""Encrypted template serialization and in-memory failed-attempt lockout.""" + +import json +import os +import threading +import time +from collections import defaultdict + +from cryptography.fernet import Fernet, InvalidToken + +from sigverify.io.schema import Signature + + +class TemplateCipher: + """Encrypt/decrypt signatures with an externally configured Fernet key.""" + + def __init__(self, key: str | bytes | None = None) -> None: + configured_key = key or os.environ.get("SIGVERIFY_FERNET_KEY") + if not configured_key: + raise RuntimeError("SIGVERIFY_FERNET_KEY must be configured before storing templates") + self._fernet = Fernet(configured_key.encode() if isinstance(configured_key, str) else configured_key) + + def encrypt(self, signature: Signature) -> bytes: + payload = { + "user_id": signature.user_id, + "label": signature.label, + "t": signature.t.tolist(), + "x": signature.x.tolist(), + "y": signature.y.tolist(), + "p": None if signature.p is None else signature.p.tolist(), + "area": None if signature.area is None else signature.area.tolist(), + "pen_down": None if signature.pen_down is None else signature.pen_down.tolist(), + "original_size": signature.original_size, + "size_normalized": signature.size_normalized, + } + return self._fernet.encrypt(json.dumps(payload).encode("utf-8")) + + def decrypt(self, token: bytes) -> Signature: + try: + data = json.loads(self._fernet.decrypt(token).decode("utf-8")) + except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError) as error: + raise ValueError("stored signature could not be decrypted") from error + return Signature(**data) + + +class AttemptLockout: + """Process-local lockout. Persistent deployments should use shared storage.""" + + def __init__(self, *, max_failures: int = 5, lockout_seconds: float = 60.0) -> None: + if max_failures < 1 or lockout_seconds <= 0: + raise ValueError("max_failures and lockout_seconds must be positive") + self.max_failures = max_failures + self.lockout_seconds = lockout_seconds + self._failures: dict[str, int] = defaultdict(int) + self._locked_until: dict[str, float] = {} + self._lock = threading.Lock() + + def remaining_lockout(self, user_id: str) -> float: + with self._lock: + return max(0.0, self._locked_until.get(user_id, 0.0) - time.monotonic()) + + def record_failure(self, user_id: str) -> None: + with self._lock: + self._failures[user_id] += 1 + if self._failures[user_id] >= self.max_failures: + self._locked_until[user_id] = time.monotonic() + self.lockout_seconds + self._failures[user_id] = 0 + + def record_success(self, user_id: str) -> None: + with self._lock: + self._failures.pop(user_id, None) + self._locked_until.pop(user_id, None) diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..7988269 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,29 @@ +import importlib + +from cryptography.fernet import Fernet +from fastapi.testclient import TestClient + + +def test_enroll_verify_and_list_users(tmp_path, monkeypatch): + database = tmp_path / "test.sqlite3" + monkeypatch.setenv("SIGVERIFY_DATABASE_URL", f"sqlite:///{database.as_posix()}") + monkeypatch.setenv("SIGVERIFY_FERNET_KEY", Fernet.generate_key().decode()) + module = importlib.import_module("app.main") + client = TestClient(module.app) + payload = { + "user_id": "test-user", + "t": [0, 1, 2, 3], + "x": [0, 1, 2, 3], + "y": [1, 2, 1, 0], + "p": [0.5, 0.6, 0.7, 0.5], + } + + assert client.get("/").status_code == 200 + assert client.post("/enroll", json=payload).status_code == 200 + assert client.post("/enroll", json=payload).status_code == 200 + verified = client.post("/verify", json=payload) + assert verified.status_code == 200 + assert verified.json()["accepted"] is True + assert client.get("/users").json()["users"] == [ + {"user_id": "test-user", "template_count": 2} + ] diff --git a/tests/test_baseline.py b/tests/test_baseline.py new file mode 100644 index 0000000..81f6ac7 --- /dev/null +++ b/tests/test_baseline.py @@ -0,0 +1,109 @@ +import numpy as np +import pytest + +from sigverify.evaluation.metrics import verification_metrics +from sigverify.features import extract_features, global_features +from sigverify.io.schema import Signature +from sigverify.io.loaders import load_mobisig +from sigverify.matching.dtw import dtw_distance +from sigverify.matching.models import DTWVerifier +from sigverify.preprocessing import preprocess_signature +from sigverify.security import AttemptLockout, TemplateCipher + + +def sample(label="genuine", scale=1.0): + t = np.array([2.0, 0.0, 1.0, 1.0, 3.0]) + x = scale * np.array([2.0, 0.0, 1.0, 99.0, 3.0]) + y = scale * np.array([0.0, 0.0, 1.0, 99.0, 1.0]) + return Signature("person-1", label, t, x, y, p=np.ones(5)) + + +def test_preprocessing_sorts_deduplicates_and_retains_original_size(): + cleaned = preprocess_signature(sample(), points=20, smoothing_sigma=0) + + assert len(cleaned.t) == 20 + assert np.all(np.diff(cleaned.t) > 0) + assert cleaned.size_normalized + assert cleaned.original_size == pytest.approx(np.hypot(3, 1)) + assert np.max(np.abs(cleaned.x)) == pytest.approx(0.5) + assert cleaned.p is not None and len(cleaned.p) == 20 + + +def test_pen_up_points_are_removed_before_resampling(): + raw = Signature( + "person-1", "genuine", [0, 1, 2], [0, 100, 2], [0, 100, 2], + pen_down=[True, False, True], + ) + cleaned = preprocess_signature(raw, points=3, smoothing_sigma=0, normalize_position=False) + assert cleaned.x.tolist() == pytest.approx([0, 1, 2]) + assert cleaned.pen_down is not None and cleaned.pen_down.all() + + +def test_features_support_optional_pressure_and_global_descriptors(): + clean = preprocess_signature(sample(), points=10, smoothing_sigma=0) + base = extract_features(clean, include_acceleration=False, include_curvature=False) + with_pressure = extract_features( + clean, include_pressure=True, include_acceleration=False, include_curvature=False + ) + + assert base.shape == (10, 5) + assert with_pressure.shape == (10, 6) + assert global_features(clean)["original_size"] > 0 + + +def test_dtw_is_zero_for_identical_sequences_and_handles_band(): + sequence = np.array([[0.0], [1.0], [2.0]]) + assert dtw_distance(sequence, sequence) == 0 + assert dtw_distance(sequence, sequence[::-1]) > 0 + assert np.isfinite(dtw_distance(sequence, sequence[:2])) + + +def test_dtw_verifier_enrollment_and_normalized_score(): + first = preprocess_signature(sample(), points=20, smoothing_sigma=0) + second = preprocess_signature(sample(scale=1.02), points=20, smoothing_sigma=0) + verifier = DTWVerifier() + verifier.enroll([first, second]) + assert verifier.score(first) < 0.1 + with pytest.raises(ValueError, match="genuine"): + verifier.enroll([first, first.copy(label="forgery")]) + + +def test_metrics_detect_perfect_separation(): + result = verification_metrics([1, 1, 0, 0], [0.1, 0.2, 0.8, 0.9]) + assert result["eer"] == 0 + + +def test_template_cipher_round_trip_and_lockout(): + from cryptography.fernet import Fernet + + signature = preprocess_signature(sample(), points=8) + cipher = TemplateCipher(Fernet.generate_key()) + restored = cipher.decrypt(cipher.encrypt(signature)) + np.testing.assert_allclose(restored.x, signature.x) + + lockout = AttemptLockout(max_failures=2, lockout_seconds=1) + lockout.record_failure("person-1") + assert lockout.remaining_lockout("person-1") == 0 + lockout.record_failure("person-1") + assert lockout.remaining_lockout("person-1") > 0 + + +def test_mobisig_loader_maps_columns_and_filename_metadata(tmp_path): + user_dir = tmp_path / "USER1" + user_dir.mkdir() + (user_dir / "SIGN_GEN_USER1_USER1_1.csv").write_text( + "x,y,timestamp,pressure,fingerarea\n1,2,10,0.5,3\n2,4,20,0.6,4\n", + encoding="utf-8", + ) + (user_dir / "SIGN_FOR_USER1_USER2_1.csv").write_text( + "x,y,timestamp,pressure,fingerarea\n1,2,10,0.5,3\n2,3,20,0.6,4\n", + encoding="utf-8", + ) + + signatures = load_mobisig(tmp_path) + + assert [(item.user_id, item.label) for item in signatures] == [ + ("USER1", "forgery"), + ("USER1", "genuine"), + ] + assert signatures[0].p is not None and signatures[0].area is not None