ObsiGate/backend/coffre_crypto.py

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