[Coffre Secrets] Phase 1a : chiffrement par enveloppe AES-256-GCM (KEK/DEK)
This commit is contained in:
@@ -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)
|
||||||
Reference in New Issue
Block a user