"""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() 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() 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)