106 lines
4.3 KiB
Python
106 lines
4.3 KiB
Python
"""Chiffrement au niveau champ pour le store local de Kazeia-central.
|
|
|
|
Choix (cf. décision 2026-06-18) : libsodium via **PyNaCl**, pas SQLCipher.
|
|
`sqlcipher3-binary` n'a de wheels que pour Linux x86_64 → impose une compilation
|
|
sur Mac/Windows, ce qui casse « facilement installable ». PyNaCl publie des wheels
|
|
`abi3` Linux/macOS/Windows (un seul couvre tout CPython ≥3.8, dont 3.14), et est
|
|
**déjà** une dépendance (déchiffrement des exports `crypto_box_seal`, §4 de la spec).
|
|
|
|
Modèle : la clé maître est dérivée du **mot de passe opérateur** (Argon2id) + un sel
|
|
aléatoire persisté. Chaque valeur sensible (texte des tours, labels patients) est
|
|
scellée indépendamment par `SecretBox` (XSalsa20-Poly1305) → seules les lignes lues
|
|
sont déchiffrées (granularité voulue, sans base entièrement chiffrée au repos).
|
|
Les métadonnées requêtables (timestamps, ids) restent en clair dans la base ; la
|
|
protection au repos de CES colonnes repose sur le chiffrement disque OS (§8 CLAUDE.md).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
import nacl.pwhash
|
|
import nacl.secret
|
|
import nacl.utils
|
|
from nacl.exceptions import CryptoError
|
|
|
|
# Paramètres KDF par défaut. MODERATE = compromis sûr/réactif pour un déverrouillage
|
|
# unique au lancement (quelques centaines de ms). Persistés en base → un changement
|
|
# futur de coût n'invalide pas les bases existantes.
|
|
_DEFAULT_OPS = nacl.pwhash.argon2id.OPSLIMIT_MODERATE
|
|
_DEFAULT_MEM = nacl.pwhash.argon2id.MEMLIMIT_MODERATE
|
|
SALT_BYTES = nacl.pwhash.argon2id.SALTBYTES # 16
|
|
KEY_BYTES = nacl.secret.SecretBox.KEY_SIZE # 32
|
|
|
|
# Sentinelle chiffrée à la création, re-déchiffrée à chaque unlock pour valider le
|
|
# mot de passe sans jamais le stocker.
|
|
_VERIFIER_PLAINTEXT = b"kazeia-central-vault-v1"
|
|
|
|
|
|
class BadPassword(Exception):
|
|
"""Mot de passe opérateur incorrect (échec de déchiffrement du vérificateur)."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class KdfParams:
|
|
salt: bytes
|
|
ops: int = _DEFAULT_OPS
|
|
mem: int = _DEFAULT_MEM
|
|
|
|
@staticmethod
|
|
def fresh(ops: int = _DEFAULT_OPS, mem: int = _DEFAULT_MEM) -> "KdfParams":
|
|
return KdfParams(salt=nacl.utils.random(SALT_BYTES), ops=ops, mem=mem)
|
|
|
|
|
|
def derive_key(password: str, p: KdfParams) -> bytes:
|
|
return nacl.pwhash.argon2id.kdf(
|
|
KEY_BYTES, password.encode("utf-8"), p.salt,
|
|
opslimit=p.ops, memlimit=p.mem,
|
|
)
|
|
|
|
|
|
class Vault:
|
|
"""Détient la clé maître dérivée. Scelle/ouvre des valeurs individuelles.
|
|
|
|
Ne jamais persister la clé ; instancier via `Vault.unlock(...)` au lancement,
|
|
garder en mémoire le temps de la session opérateur.
|
|
"""
|
|
|
|
def __init__(self, key: bytes) -> None:
|
|
self._box = nacl.secret.SecretBox(key)
|
|
|
|
# ---- cycle de vie -----------------------------------------------------
|
|
@classmethod
|
|
def setup(cls, password: str, ops: int = _DEFAULT_OPS,
|
|
mem: int = _DEFAULT_MEM) -> tuple["Vault", KdfParams, bytes]:
|
|
"""Première initialisation : dérive la clé, renvoie le vault, les params
|
|
KDF à persister et le blob vérificateur à stocker. `ops`/`mem` réglables
|
|
(défaut prod MODERATE ; tests peuvent baisser pour la vitesse)."""
|
|
params = KdfParams.fresh(ops, mem)
|
|
v = cls(derive_key(password, params))
|
|
return v, params, v.seal_bytes(_VERIFIER_PLAINTEXT)
|
|
|
|
@classmethod
|
|
def unlock(cls, password: str, params: KdfParams, verifier: bytes) -> "Vault":
|
|
"""Ré-ouverture : dérive la clé et valide contre le vérificateur stocké."""
|
|
v = cls(derive_key(password, params))
|
|
try:
|
|
if v.open_bytes(verifier) != _VERIFIER_PLAINTEXT:
|
|
raise BadPassword()
|
|
except CryptoError as e:
|
|
raise BadPassword() from e
|
|
return v
|
|
|
|
# ---- scellage ---------------------------------------------------------
|
|
def seal_bytes(self, data: bytes) -> bytes:
|
|
# SecretBox.encrypt préfixe un nonce aléatoire → blob auto-suffisant.
|
|
return bytes(self._box.encrypt(data))
|
|
|
|
def open_bytes(self, blob: bytes) -> bytes:
|
|
return self._box.decrypt(blob)
|
|
|
|
def seal(self, text: str | None) -> bytes | None:
|
|
return None if text is None else self.seal_bytes(text.encode("utf-8"))
|
|
|
|
def open(self, blob: bytes | None) -> str | None:
|
|
return None if blob is None else self.open_bytes(blob).decode("utf-8")
|