"""Store local chiffré de Kazeia-central. `sqlite3` (stdlib, zéro install) + chiffrement au niveau champ (`crypto.Vault`, PyNaCl). Contient l'archive clinique rapatriée (tours de conversation, PII de santé chiffrée), le mapping `serial → label patient` (§9), et la piste d'audit (§8). Posture : - Texte des tours et labels patients → **chiffrés** (colonnes `*_enc BLOB`). - Métadonnées (serial, profile_id, session_id, timestamps) → **clair**, pour rester requêtables (dashboard, collecte incrémentale, fan-out). Protégées au repos par le chiffrement disque OS (§8). - Tout accès sensible est journalisé (`audit`). Le store s'ouvre verrouillé ; `unlock(password)` dérive la clé et la garde en mémoire pour la session. Sans unlock, seules les métadonnées sont lisibles, jamais le contenu. Thread-safety : les endpoints FastAPI synchrones tournent dans un threadpool → connexion ouverte avec `check_same_thread=False` et accès sérialisé par un `RLock`. """ from __future__ import annotations import json import sqlite3 import threading from pathlib import Path from typing import Any, Iterable from .crypto import KdfParams, Vault _SCHEMA = """ CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value BLOB ); CREATE TABLE IF NOT EXISTS devices ( serial TEXT PRIMARY KEY, label_enc BLOB, -- label patient, chiffré created_at INTEGER NOT NULL, last_seen_at INTEGER ); CREATE TABLE IF NOT EXISTS turns ( serial TEXT NOT NULL, profile_id TEXT, session_id TEXT NOT NULL, turn_id TEXT NOT NULL, -- `id` du tour côté provider ts INTEGER, -- timestamp du tour (clair → tri/incrémental) role TEXT, -- PATIENT | KAZEIA text_enc BLOB, -- contenu, chiffré (PII) ttft_ms INTEGER, total_ms INTEGER, voice_used TEXT, model_used TEXT, archived_at INTEGER NOT NULL, PRIMARY KEY (serial, session_id, turn_id) -- dédoublonne les re-pulls ); CREATE INDEX IF NOT EXISTS idx_turns_session ON turns(serial, session_id, ts); CREATE INDEX IF NOT EXISTS idx_turns_profile ON turns(serial, profile_id, ts); CREATE TABLE IF NOT EXISTS audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, actor TEXT, action TEXT NOT NULL, target TEXT, detail TEXT -- métadonnée d'audit (pas de PII) → clair ); """ class StoreLocked(Exception): """Opération sur du contenu chiffré alors que le store n'est pas déverrouillé.""" class Store: def __init__(self, path: str | Path, *, kdf_ops: int | None = None, kdf_mem: int | None = None) -> None: # kdf_ops/kdf_mem : coût Argon2id à l'initialisation (1ᵉʳ unlock). None = # défaut prod MODERATE. Les tests les baissent pour la vitesse ; ils sont # persistés, donc les ré-ouvertures réutilisent les mêmes. self.path = str(path) self._kdf_ops = kdf_ops self._kdf_mem = kdf_mem self._lock = threading.RLock() self._db = sqlite3.connect(self.path, check_same_thread=False) self._db.row_factory = sqlite3.Row self._db.execute("PRAGMA journal_mode=WAL") self._db.execute("PRAGMA foreign_keys=ON") self._db.executescript(_SCHEMA) self._db.commit() self._vault: Vault | None = None def close(self) -> None: with self._lock: self._db.close() def __enter__(self) -> "Store": return self def __exit__(self, *exc: object) -> None: self.close() # ---- meta ------------------------------------------------------------- def _get_meta(self, key: str) -> bytes | None: row = self._db.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() return row["value"] if row else None def _set_meta(self, key: str, value: bytes) -> None: self._db.execute( "INSERT INTO meta(key, value) VALUES(?, ?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value), ) self._db.commit() @property def is_initialized(self) -> bool: with self._lock: return self._get_meta("kdf_salt") is not None @property def is_unlocked(self) -> bool: return self._vault is not None # ---- déverrouillage --------------------------------------------------- def unlock(self, password: str) -> None: """Initialise le coffre au 1ᵉʳ appel, sinon valide le mot de passe. Lève `crypto.BadPassword` si incorrect.""" with self._lock: if not self._get_meta("kdf_salt"): from .crypto import _DEFAULT_MEM, _DEFAULT_OPS vault, params, verifier = Vault.setup( password, ops=self._kdf_ops if self._kdf_ops is not None else _DEFAULT_OPS, mem=self._kdf_mem if self._kdf_mem is not None else _DEFAULT_MEM, ) self._set_meta("kdf_salt", params.salt) self._set_meta("kdf_ops", str(params.ops).encode()) self._set_meta("kdf_mem", str(params.mem).encode()) self._set_meta("verifier", verifier) self._vault = vault return params = KdfParams( salt=self._get_meta("kdf_salt"), ops=int(self._get_meta("kdf_ops")), mem=int(self._get_meta("kdf_mem")), ) self._vault = Vault.unlock(password, params, self._get_meta("verifier")) def lock(self) -> None: """Oublie la clé en mémoire (re-verrouille sans fermer la base).""" self._vault = None def _require_vault(self) -> Vault: if self._vault is None: raise StoreLocked("store verrouillé : appeler unlock(password) d'abord") return self._vault # ---- mapping serial → label patient (§9) ------------------------------ def set_device_label(self, serial: str, label: str, *, now: int) -> None: with self._lock: v = self._require_vault() self._db.execute( "INSERT INTO devices(serial, label_enc, created_at, last_seen_at) " "VALUES(?,?,?,?) ON CONFLICT(serial) DO UPDATE SET " "label_enc=excluded.label_enc, last_seen_at=excluded.last_seen_at", (serial, v.seal(label), now, now), ) self._db.commit() def device_label(self, serial: str) -> str | None: with self._lock: v = self._require_vault() row = self._db.execute( "SELECT label_enc FROM devices WHERE serial=?", (serial,) ).fetchone() return v.open(row["label_enc"]) if row else None def device_labels(self) -> dict[str, str | None]: """Tous les labels {serial: label} — pour enrichir la liste du parc.""" with self._lock: v = self._require_vault() return {r["serial"]: v.open(r["label_enc"]) for r in self._db.execute("SELECT serial, label_enc FROM devices")} def devices(self) -> list[dict[str, Any]]: with self._lock: v = self._require_vault() return [{ "serial": r["serial"], "label": v.open(r["label_enc"]), "created_at": r["created_at"], "last_seen_at": r["last_seen_at"], } for r in self._db.execute("SELECT * FROM devices ORDER BY created_at")] # ---- archive des tours (collecte) ------------------------------------- def archive_turns(self, serial: str, turns: Iterable[dict[str, Any]], *, now: int) -> int: """Insère/ignore des tours (idempotent par (serial, session_id, turn_id)). Chaque `text` est chiffré. Renvoie le nombre de NOUVEAUX tours archivés.""" with self._lock: v = self._require_vault() before = self._db.total_changes self._db.executemany( "INSERT OR IGNORE INTO turns(serial, profile_id, session_id, turn_id, ts, " "role, text_enc, ttft_ms, total_ms, voice_used, model_used, archived_at) " "VALUES(:serial,:profile_id,:session_id,:turn_id,:ts,:role,:text_enc," ":ttft_ms,:total_ms,:voice_used,:model_used,:archived_at)", [{ "serial": serial, "profile_id": t.get("profile_id"), "session_id": t["session_id"], "turn_id": str(t.get("id", t.get("turn_id"))), "ts": t.get("timestamp", t.get("ts")), "role": t.get("role"), "text_enc": v.seal(t.get("text")), "ttft_ms": t.get("ttft_ms"), "total_ms": t.get("total_ms"), "voice_used": t.get("voice_used"), "model_used": t.get("model_used"), "archived_at": now, } for t in turns], ) self._db.commit() return self._db.total_changes - before def session_turns(self, serial: str, session_id: str) -> list[dict[str, Any]]: with self._lock: v = self._require_vault() rows = self._db.execute( "SELECT * FROM turns WHERE serial=? AND session_id=? ORDER BY ts, turn_id", (serial, session_id), ).fetchall() out = [] for r in rows: d = dict(r) d["text"] = v.open(d.pop("text_enc")) out.append(d) return out def last_turn_ts(self, serial: str, session_id: str | None = None) -> int | None: """Dernier timestamp archivé → borne `since` pour la collecte incrémentale (autonomie). Pas besoin du vault (métadonnée en clair).""" with self._lock: if session_id is None: row = self._db.execute( "SELECT MAX(ts) m FROM turns WHERE serial=?", (serial,) ).fetchone() else: row = self._db.execute( "SELECT MAX(ts) m FROM turns WHERE serial=? AND session_id=?", (serial, session_id), ).fetchone() return row["m"] if row else None # ---- audit (§8) ------------------------------------------------------- def audit(self, action: str, *, actor: str | None = None, target: str | None = None, detail: dict[str, Any] | None = None, now: int) -> None: with self._lock: self._db.execute( "INSERT INTO audit(ts, actor, action, target, detail) VALUES(?,?,?,?,?)", (now, actor, action, target, json.dumps(detail) if detail else None), ) self._db.commit() def audit_log(self, limit: int = 200) -> list[dict[str, Any]]: with self._lock: return [dict(r) for r in self._db.execute( "SELECT * FROM audit ORDER BY id DESC LIMIT ?", (limit,) )]