From ff1ccca3cd83bc4a5d8f258f6c351bfc2aa1e2a5 Mon Sep 17 00:00:00 2001 From: bruno Date: Tue, 27 Oct 2026 21:45:32 -0400 Subject: [PATCH] [Coffre Secrets] Phase 1a : chiffrement par enveloppe AES-256-GCM (KEK/DEK) --- backend/coffre_crypto.py | 204 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) create mode 100644 backend/coffre_crypto.py diff --git a/backend/coffre_crypto.py b/backend/coffre_crypto.py new file mode 100644 index 0000000..438773b --- /dev/null +++ b/backend/coffre_crypto.py @@ -0,0 +1,204 @@ +"""Chiffrement par enveloppe du coffre Secrets ObsiGate (Phase 1a). + +Principe (spec §5) : deux niveaux de clés, jamais de valeur en clair au repos. + +- KEK : clé maître 256 bits, fournie par l'opérateur via ``OBSIGATE_KEK`` + (base64, 32 octets) ou ``OBSIGATE_KEK_FILE`` (fichier 0600, hors dépôt, + hors arborescence des vaults). Elle n'est JAMAIS dérivée du secret + d'application, jamais journalisée, jamais renvoyée par API. +- DEK : une clé aléatoire par coffre, stockée uniquement chiffrée par la KEK. +- Version de secret : ciphertext AES-256-GCM + nonce unique + tag GCM + (le tag est inclus dans le ciphertext renvoyé par ``AESGCM.encrypt``). + Un tag invalide ou un fichier altéré = refus, jamais de valeur fausse. + +Le module reste importable sans ``cryptography`` installé : les opérations +échouent alors avec ``CryptoIndisponibleError`` au lieu de casser le +démarrage de l'application (le routeur répond 503 dans ce cas). +""" + +from __future__ import annotations + +import base64 +import binascii +import logging +import os +import stat +from pathlib import Path + +logger = logging.getLogger("obsigate.coffre_crypto") + +try: # cryptography est une dépendance de la Phase 1a (requirements.txt) + from cryptography.exceptions import InvalidTag + from cryptography.hazmat.primitives.ciphers.aead import AESGCM +except Exception: # pragma: no cover - environnement sans cryptography + AESGCM = None # type: ignore[assignment] + InvalidTag = Exception # type: ignore[assignment,misc] + + +class CoffreCryptoError(Exception): + """Erreur générique du chiffrement du coffre.""" + + +class KEKAbsenteError(CoffreCryptoError): + """Aucune KEK configurée (ni OBSIGATE_KEK, ni OBSIGATE_KEK_FILE).""" + + +class KEKInvalideError(CoffreCryptoError): + """KEK mal formée (ni base64 32 octets, ni hex 64 caractères).""" + + +class TagInvalideError(CoffreCryptoError): + """Authentification GCM échouée : données altérées ou mauvaise clé.""" + + +class CryptoIndisponibleError(CoffreCryptoError): + """La bibliothèque ``cryptography`` n'est pas installée.""" + + +def _require_aesgcm() -> None: + if AESGCM is None: + raise CryptoIndisponibleError( + "cryptography n'est pas installé (dépendance Phase 1a)" + ) + + +def parse_kek(raw: str) -> bytes: + """Parse une KEK : base64 (32 octets) ou hex (64 caractères). + + Accepte le base64 standard et URL-safe, avec ou sans padding. + """ + value = (raw or "").strip() + if not value: + raise KEKInvalideError("KEK vide") + # Hex d'abord si la forme correspond (64 caractères hex). + if len(value) == 64: + try: + decoded = bytes.fromhex(value) + except ValueError: + decoded = b"" + if len(decoded) == 32: + return decoded + padded = value + "=" * (-len(value) % 4) + for decoder in (base64.urlsafe_b64decode, base64.b64decode): + try: + decoded = decoder(padded) + except (binascii.Error, ValueError): + continue + if len(decoded) == 32: + return decoded + raise KEKInvalideError("KEK invalide : attendu base64 de 32 octets") + + +def generate_kek_b64() -> str: + """Génère une KEK 256 bits, encodée base64 (à sauvegarder hors dépôt).""" + return base64.b64encode(os.urandom(32)).decode("ascii") + + +def load_kek() -> bytes: + """Charge la KEK depuis l'environnement ou le fichier dédié. + + Priorité : ``OBSIGATE_KEK`` puis ``OBSIGATE_KEK_FILE``. Le fichier doit + être en 0600 (aucun droit groupe/autre) ; sinon, refus explicite. + """ + env_kek = os.environ.get("OBSIGATE_KEK", "").strip() + if env_kek: + return parse_kek(env_kek) + kek_file = os.environ.get("OBSIGATE_KEK_FILE", "").strip() + if kek_file: + path = Path(kek_file) + try: + mode = path.stat().st_mode + except OSError as exc: + raise KEKAbsenteError(f"Fichier KEK illisible : {kek_file}") from exc + if mode & (stat.S_IRWXG | stat.S_IRWXO): + raise KEKInvalideError( + "Fichier KEK trop permissif : attendu 0600, " + f"droits groupe/autre détectés sur {kek_file}" + ) + try: + return parse_kek(path.read_text(encoding="utf-8")) + except OSError as exc: + raise KEKAbsenteError(f"Fichier KEK illisible : {kek_file}") from exc + raise KEKAbsenteError("Aucune KEK configurée (OBSIGATE_KEK / OBSIGATE_KEK_FILE)") + + +def kek_presente() -> bool: + """True si une KEK valide est chargeable, False sinon (jamais d'exception).""" + try: + load_kek() + except CoffreCryptoError: + return False + return True + + +def generate_dek() -> bytes: + """Génère une DEK aléatoire 256 bits pour un coffre.""" + return os.urandom(32) + + +def _b64(data: bytes) -> str: + return base64.b64encode(data).decode("ascii") + + +def _unb64(value: str) -> bytes: + return base64.b64decode(value) + + +def encrypt_with_key(key: bytes, plaintext: bytes, aad: bytes) -> dict: + """Chiffre avec AES-256-GCM. Retourne {nonce, ciphertext} en base64.""" + _require_aesgcm() + nonce = os.urandom(12) + ciphertext = AESGCM(key).encrypt(nonce, plaintext, aad) + return {"nonce": _b64(nonce), "ciphertext": _b64(ciphertext)} + + +def decrypt_with_key(key: bytes, nonce_b64: str, ciphertext_b64: str, aad: bytes) -> bytes: + """Déchiffre avec AES-256-GCM. Tag invalide -> TagInvalideError.""" + _require_aesgcm() + try: + return AESGCM(key).decrypt(_unb64(nonce_b64), _unb64(ciphertext_b64), aad) + except InvalidTag as exc: + raise TagInvalideError("Tag GCM invalide : données altérées ou mauvaise clé") from exc + except (binascii.Error, ValueError) as exc: + raise TagInvalideError("Données chiffrées mal formées") from exc + + +def wrap_dek(dek: bytes, kek: bytes, coffre_id: str) -> dict: + """Chiffre une DEK avec la KEK (AAD = identifiant du coffre).""" + return encrypt_with_key(kek, dek, coffre_id.encode("utf-8")) + + +def unwrap_dek(wrapped: dict, kek: bytes, coffre_id: str) -> bytes: + """Déchiffre une DEK avec la KEK.""" + return decrypt_with_key( + kek, wrapped["nonce"], wrapped["ciphertext"], coffre_id.encode("utf-8") + ) + + +def encrypt_secret_value(dek: bytes, secret_id: str, version: int, valeur: str) -> dict: + """Chiffre une valeur de secret (AAD = secret_id:version).""" + aad = f"{secret_id}:{version}".encode("utf-8") + return encrypt_with_key(dek, valeur.encode("utf-8"), aad) + + +def decrypt_secret_value( + dek: bytes, secret_id: str, version: int, nonce_b64: str, ciphertext_b64: str +) -> str: + """Déchiffre une valeur de secret.""" + aad = f"{secret_id}:{version}".encode("utf-8") + return decrypt_with_key(dek, nonce_b64, ciphertext_b64, aad).decode("utf-8") + + +def reencrypt_secret_value( + ancienne_dek: bytes, + nouvelle_dek: bytes, + secret_id: str, + version: int, + nonce_b64: str, + ciphertext_b64: str, +) -> dict: + """Re-chiffre une version avec une nouvelle DEK (rotation DEK).""" + valeur = decrypt_secret_value( + ancienne_dek, secret_id, version, nonce_b64, ciphertext_b64 + ) + return encrypt_secret_value(nouvelle_dek, secret_id, version, valeur)