[Coffre Secrets] Phase 1a : store coffres/valeurs/octrois/consommateurs (0600, atomique, RLock)

backend/coffre_store.py

@@ -0,0 +1,860 @@
+"""Store du coffre Secrets ObsiGate (Phase 1a).
+
+Patron repris de ``backend/tools/secrets.py`` : JSON dans ``data/``,
+écriture atomique (tmp + rename), verrou ``RLock`` sur les cycles
+lecture-modification-écriture, répertoire honorant ``OBSIGATE_DATA_DIR``.
+
+Fichiers (permissions 0600) :
+- ``data/coffres.json`` : coffres (métadonnées + DEK chiffrées),
+  consommateurs (jetons hachés SHA-256 seulement, jamais en clair),
+  octrois (références, statuts, expirations).
+- ``data/coffres_valeurs.json`` : versions chiffrées des secrets.
+
+Aucune valeur en clair n'est jamais persistée ici, et aucune fonction de
+ce module ne journalise de valeur ni de texte chiffré.
+
+Modèle :
+- Coffre : id, slug, nom, description, proprietaire, perimetre
+  (personnel | partage | instance), dek_chiffree (ou None en mode
+  dégradé sans KEK), statut (actif | archive), dates.
+- Secret : id, coffre_id, nom, type, service, note, expiration,
+  version_courante, cree_le, modifie_le, derniere_utilisation.
+- Version : numero, nonce, ciphertext, cree_le, remplacee_le.
+  Seule la version courante est distribuable.
+- Consommateur : id (sert de jti), nom, jeton_hash, jeton_prefixe,
+  statut (actif | revoque), cree_par, cree_le.
+- Octroi : id, coffre_id, secret_id (None = tout le coffre),
+  cible_type (consommateur | utilisateur | executeur | workspace),
+  cible_id, expiration, statut (actif | revoque | expire),
+  cree_par, cree_le, revoque_le.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import logging
+import os
+import re
+import secrets as _secrets
+import threading
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from backend.coffre_crypto import (
+    KEKAbsenteError,
+    generate_dek,
+    load_kek,
+    parse_kek,
+    reencrypt_secret_value,
+    unwrap_dek,
+    wrap_dek,
+    encrypt_secret_value,
+    decrypt_secret_value,
+)
+
+logger = logging.getLogger("obsigate.coffre_store")
+
+_lock = threading.RLock()
+
+SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{2,63}$")
+NOM_SECRET_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{2,127}$")
+TYPES_SECRET = {"api_key", "token", "password", "certificate", "autre"}
+PERIMETRES = {"personnel", "partage", "instance"}
+CIBLES_OCTROI = {"consommateur", "utilisateur", "executeur", "workspace"}
+
+MAX_VALEUR = 65536
+MAX_TEXTE = 2000
+DUREE_OCTROI_DEFAUT_JOURS = 90
+
+# Motifs refusés dans les champs de métadonnées (description, note) :
+# une valeur ressemblant à un secret ne doit jamais transiter par un champ
+# qui, lui, est renvoyé en clair dans les métadonnées.
+_SECRET_LIKE_PATTERNS = (
+    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
+    re.compile(r"\bsk-[A-Za-z0-9]{8,}\b"),
+    re.compile(r"\bgh[opsu]_[A-Za-z0-9]{8,}\b"),
+    re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
+    re.compile(
+        r"(?i)\b(token|secret|mot_de_passe|password|api[-_]?key)\s*[:=]\s*\S+"
+    ),
+)
+
+
+class StoreError(Exception):
+    """Erreur générique du store."""
+
+
+class CoffreExisteError(StoreError):
+    pass
+
+
+class SecretExisteError(StoreError):
+    pass
+
+
+class IntrouvableError(StoreError):
+    pass
+
+
+class ValidationError(StoreError):
+    pass
+
+
+class SecretExpireError(StoreError):
+    pass
+
+
+def resolve_data_dir(data_dir: str | Path | None = None) -> Path:
+    """Répertoire de persistance (OBSIGATE_DATA_DIR, défaut ``data``)."""
+    if data_dir is not None:
+        return Path(data_dir)
+    return Path(os.environ.get("OBSIGATE_DATA_DIR", "data"))
+
+
+def _coffres_path(base: Path) -> Path:
+    return base / "coffres.json"
+
+
+def _valeurs_path(base: Path) -> Path:
+    return base / "coffres_valeurs.json"
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _parse_date(value: str | None) -> datetime | None:
+    """Parse une date ISO (datetime) ou une date seule (fin de journée UTC)."""
+    if value is None:
+        return None
+    text = str(value).strip()
+    if not text:
+        return None
+    if len(text) == 10:
+        text = text + "T23:59:59+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError as exc:
+        raise ValidationError(f"Date invalide : {value!r}") from exc
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def est_expire(expiration: str | None, maintenant: datetime | None = None) -> bool:
+    """True si la date d'expiration est passée (None = jamais)."""
+    parsed = _parse_date(expiration)
+    if parsed is None:
+        return False
+    ref = maintenant or datetime.now(timezone.utc)
+    return parsed <= ref
+
+
+def _read_json(path: Path, default: dict) -> dict:
+    if not path.exists():
+        return json.loads(json.dumps(default))
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        logger.warning("Store coffre illisible (%s) : %s", path, exc)
+        return json.loads(json.dumps(default))
+    return data if isinstance(data, dict) else json.loads(json.dumps(default))
+
+
+def _write_json(path: Path, data: dict) -> None:
+    """Écriture atomique tmp+rename, fichier final en 0600."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+    try:
+        os.chmod(tmp, 0o600)
+    except OSError:
+        pass
+    os.replace(tmp, path)
+    try:
+        os.chmod(path, 0o600)
+    except OSError:
+        pass
+
+
+def _load_coffres(base: Path) -> dict:
+    return _read_json(
+        _coffres_path(base),
+        {"coffres": [], "consommateurs": [], "octrois": []},
+    )
+
+
+def _load_valeurs(base: Path) -> dict:
+    return _read_json(_valeurs_path(base), {"versions": {}})
+
+
+def _save_coffres(base: Path, data: dict) -> None:
+    _write_json(_coffres_path(base), data)
+
+
+def _save_valeurs(base: Path, data: dict) -> None:
+    _write_json(_valeurs_path(base), data)
+
+
+def valider_slug(slug: str) -> str:
+    value = (slug or "").strip().lower()
+    if not SLUG_RE.match(value):
+        raise ValidationError(
+            "Slug invalide : 3 à 64 caractères, minuscules, chiffres, - ou _"
+        )
+    return value
+
+
+def valider_nom_secret(nom: str) -> str:
+    value = (nom or "").strip().lower()
+    if not NOM_SECRET_RE.match(value):
+        raise ValidationError(
+            "Nom de secret invalide : 3 à 128 caractères, minuscules, chiffres, - ou _"
+        )
+    return value
+
+
+def refuser_valeur_dans_metadonnees(texte: str | None, champ: str) -> None:
+    """Refuse une valeur ressemblant à un secret dans un champ métadonnée."""
+    if not texte:
+        return
+    for motif in _SECRET_LIKE_PATTERNS:
+        if motif.search(texte):
+            raise ValidationError(
+                f"Le champ {champ} ressemble à une valeur de secret ; "
+                "utilisez le champ valeur du secret, jamais une métadonnée"
+            )
+
+
+def _valider_texte(texte: str | None, champ: str) -> str:
+    value = (texte or "").strip()
+    if len(value) > MAX_TEXTE:
+        raise ValidationError(f"Le champ {champ} dépasse {MAX_TEXTE} caractères")
+    refuser_valeur_dans_metadonnees(value, champ)
+    return value
+
+
+# ---------------------------------------------------------------------------
+# Coffres
+# ---------------------------------------------------------------------------
+
+def list_coffres(data_dir: str | Path | None = None) -> list:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        return list(_load_coffres(base)["coffres"])
+
+
+def get_coffre(slug: str, data_dir: str | Path | None = None) -> dict | None:
+    base = resolve_data_dir(data_dir)
+    slug = valider_slug(slug)
+    with _lock:
+        for coffre in _load_coffres(base)["coffres"]:
+            if coffre["slug"] == slug:
+                return coffre
+    return None
+
+
+def get_coffre_or_404(slug: str, data_dir: str | Path | None = None) -> dict:
+    coffre = get_coffre(slug, data_dir)
+    if coffre is None:
+        raise IntrouvableError(f"Coffre introuvable : {slug}")
+    return coffre
+
+
+def get_coffre_by_id(coffre_id: str, data_dir: str | Path | None = None) -> dict | None:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        for coffre in _load_coffres(base)["coffres"]:
+            if coffre["id"] == coffre_id:
+                return coffre
+    return None
+
+
+def get_octroi(octroi_id: str, data_dir: str | Path | None = None) -> dict | None:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        for octroi in _load_coffres(base)["octrois"]:
+            if octroi["id"] == octroi_id:
+                return octroi
+    return None
+
+
+def create_coffre(
+    slug: str,
+    nom: str,
+    proprietaire: str,
+    perimetre: str = "personnel",
+    description: str = "",
+    data_dir: str | Path | None = None,
+) -> dict:
+    """Crée un coffre. Sans KEK, la DEK reste None (mode dégradé) : les
+    métadonnées sont créées, mais aucun secret ne pourra être ajouté tant
+    que la KEK n'est pas configurée."""
+    base = resolve_data_dir(data_dir)
+    slug = valider_slug(slug)
+    if perimetre not in PERIMETRES:
+        raise ValidationError(f"Périmètre invalide : {perimetre}")
+    if not (proprietaire or "").strip():
+        raise ValidationError("Propriétaire requis")
+    description = _valider_texte(description, "description")
+    nom = (nom or "").strip()
+    if not nom or len(nom) > 128:
+        raise ValidationError("Nom de coffre requis (128 caractères max)")
+    with _lock:
+        data = _load_coffres(base)
+        if any(c["slug"] == slug for c in data["coffres"]):
+            raise CoffreExisteError(f"Coffre déjà existant : {slug}")
+        coffre = {
+            "id": str(uuid.uuid4()),
+            "slug": slug,
+            "nom": nom,
+            "description": description,
+            "proprietaire": proprietaire.strip(),
+            "perimetre": perimetre,
+            "dek_chiffree": None,
+            "statut": "actif",
+            "cree_le": _now_iso(),
+        }
+        try:
+            kek = load_kek()
+        except Exception:
+            kek = None
+        if kek is not None:
+            coffre["dek_chiffree"] = wrap_dek(generate_dek(), kek, coffre["id"])
+        data["coffres"].append(coffre)
+        _save_coffres(base, data)
+        return coffre
+
+
+def ensure_dek(coffre: dict, data_dir: str | Path | None = None) -> bytes:
+    """Retourne la DEK en clair (mémoire seulement). Génère et persiste la
+    DEK si le coffre a été créé en mode dégradé et que la KEK est là."""
+    base = resolve_data_dir(data_dir)
+    kek = load_kek()  # lève KEKAbsenteError si aucune KEK
+    if coffre.get("dek_chiffree"):
+        return unwrap_dek(coffre["dek_chiffree"], kek, coffre["id"])
+    with _lock:
+        data = _load_coffres(base)
+        for item in data["coffres"]:
+            if item["id"] == coffre["id"]:
+                if item.get("dek_chiffree"):
+                    return unwrap_dek(item["dek_chiffree"], kek, item["id"])
+                dek = generate_dek()
+                item["dek_chiffree"] = wrap_dek(dek, kek, item["id"])
+                _save_coffres(base, data)
+                return dek
+    raise IntrouvableError("Coffre introuvable pour génération de DEK")
+
+
+# Les secrets vivent dans coffres.json (métadonnées), leurs versions dans
+# coffres_valeurs.json (chiffrées). Les deux chargements sont faits ici
+# pour garder une seule source de vérité dans ce module.
+def _load_tout(base: Path) -> tuple[dict, dict]:
+    data = _load_coffres(base)
+    if "secrets" not in data:
+        data["secrets"] = []
+    return data, _load_valeurs(base)
+
+
+def compter_secrets(coffre_id: str, data_dir: str | Path | None = None) -> int:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        data, _ = _load_tout(base)
+        return sum(1 for s in data["secrets"] if s["coffre_id"] == coffre_id)
+
+
+def list_secrets(coffre_id: str, data_dir: str | Path | None = None) -> list:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        data, _ = _load_tout(base)
+        return [s for s in data["secrets"] if s["coffre_id"] == coffre_id]
+
+
+def get_secret(
+    coffre_id: str, nom: str, data_dir: str | Path | None = None
+) -> dict | None:
+    base = resolve_data_dir(data_dir)
+    nom = valider_nom_secret(nom)
+    with _lock:
+        data, _ = _load_tout(base)
+        for secret in data["secrets"]:
+            if secret["coffre_id"] == coffre_id and secret["nom"] == nom:
+                return secret
+    return None
+
+
+def secret_metadata(secret: dict, octrois_actifs: int = 0) -> dict:
+    """Vue métadonnées seules d'un secret : jamais de valeur, jamais de
+    texte chiffré, jamais de nonce."""
+    return {
+        "id": secret["id"],
+        "nom": secret["nom"],
+        "type": secret["type"],
+        "service": secret.get("service", ""),
+        "note": secret.get("note", ""),
+        "expiration": secret.get("expiration"),
+        "expire": est_expire(secret.get("expiration")),
+        "version": secret["version_courante"],
+        "has_value": True,
+        "cree_le": secret["cree_le"],
+        "modifie_le": secret["modifie_le"],
+        "derniere_utilisation": secret.get("derniere_utilisation"),
+        "octrois_actifs": octrois_actifs,
+        "reference": f"coffre:{secret['coffre_slug']}/{secret['nom']}",
+    }
+
+
+def create_secret(
+    coffre: dict,
+    nom: str,
+    type_secret: str,
+    valeur: str,
+    service: str = "",
+    expiration: str | None = None,
+    note: str = "",
+    data_dir: str | Path | None = None,
+) -> dict:
+    """Chiffre et stocke un secret (nouvelle version 1)."""
+    base = resolve_data_dir(data_dir)
+    nom = valider_nom_secret(nom)
+    if type_secret not in TYPES_SECRET:
+        raise ValidationError(f"Type de secret invalide : {type_secret}")
+    if not valeur or len(valeur) > MAX_VALEUR:
+        raise ValidationError("Valeur requise (64 KiB max)")
+    service = (service or "").strip()
+    if len(service) > 128:
+        raise ValidationError("Service trop long (128 caractères max)")
+    note = _valider_texte(note, "note")
+    if expiration is not None:
+        _parse_date(expiration)
+    dek = ensure_dek(coffre, base)  # KEKAbsenteError si pas de KEK
+    with _lock:
+        data, valeurs = _load_tout(base)
+        if any(
+            s["coffre_id"] == coffre["id"] and s["nom"] == nom
+            for s in data["secrets"]
+        ):
+            raise SecretExisteError(f"Secret déjà existant : {nom}")
+        secret_id = str(uuid.uuid4())
+        chiffre = encrypt_secret_value(dek, secret_id, 1, valeur)
+        secret = {
+            "id": secret_id,
+            "coffre_id": coffre["id"],
+            "coffre_slug": coffre["slug"],
+            "nom": nom,
+            "type": type_secret,
+            "service": service,
+            "note": note,
+            "expiration": expiration,
+            "version_courante": 1,
+            "cree_le": _now_iso(),
+            "modifie_le": _now_iso(),
+            "derniere_utilisation": None,
+        }
+        data["secrets"].append(secret)
+        valeurs["versions"][secret_id] = [
+            {
+                "numero": 1,
+                "nonce": chiffre["nonce"],
+                "ciphertext": chiffre["ciphertext"],
+                "cree_le": secret["cree_le"],
+                "remplacee_le": None,
+            }
+        ]
+        _save_coffres(base, data)
+        _save_valeurs(base, valeurs)
+        return secret
+
+
+def rotate_secret(
+    coffre: dict, nom: str, valeur: str, data_dir: str | Path | None = None
+) -> dict:
+    """Rotation : nouvelle version active immédiatement, l'ancienne n'est
+    plus distribuable (seule la version courante est lue à la distribution)."""
+    base = resolve_data_dir(data_dir)
+    nom = valider_nom_secret(nom)
+    if not valeur or len(valeur) > MAX_VALEUR:
+        raise ValidationError("Valeur requise (64 KiB max)")
+    dek = ensure_dek(coffre, base)
+    with _lock:
+        data, valeurs = _load_tout(base)
+        secret = None
+        for item in data["secrets"]:
+            if item["coffre_id"] == coffre["id"] and item["nom"] == nom:
+                secret = item
+                break
+        if secret is None:
+            raise IntrouvableError(f"Secret introuvable : {nom}")
+        nouvelle_version = secret["version_courante"] + 1
+        chiffre = encrypt_secret_value(dek, secret["id"], nouvelle_version, valeur)
+        versions = valeurs["versions"].setdefault(secret["id"], [])
+        for version in versions:
+            if version["remplacee_le"] is None:
+                version["remplacee_le"] = _now_iso()
+        versions.append(
+            {
+                "numero": nouvelle_version,
+                "nonce": chiffre["nonce"],
+                "ciphertext": chiffre["ciphertext"],
+                "cree_le": _now_iso(),
+                "remplacee_le": None,
+            }
+        )
+        secret["version_courante"] = nouvelle_version
+        secret["modifie_le"] = _now_iso()
+        _save_coffres(base, data)
+        _save_valeurs(base, valeurs)
+        return secret
+
+
+def delete_secret(
+    coffre: dict, nom: str, data_dir: str | Path | None = None
+) -> tuple[dict, list]:
+    """Supprime un secret, ses versions et ses octrois (cascade)."""
+    base = resolve_data_dir(data_dir)
+    nom = valider_nom_secret(nom)
+    with _lock:
+        data, valeurs = _load_tout(base)
+        secret = None
+        for item in data["secrets"]:
+            if item["coffre_id"] == coffre["id"] and item["nom"] == nom:
+                secret = item
+                break
+        if secret is None:
+            raise IntrouvableError(f"Secret introuvable : {nom}")
+        data["secrets"] = [s for s in data["secrets"] if s["id"] != secret["id"]]
+        valeurs["versions"].pop(secret["id"], None)
+        octrois_supprimes = [
+            o for o in data["octrois"] if o.get("secret_id") == secret["id"]
+        ]
+        data["octrois"] = [
+            o for o in data["octrois"] if o.get("secret_id") != secret["id"]
+        ]
+        _save_coffres(base, data)
+        _save_valeurs(base, valeurs)
+        return secret, octrois_supprimes
+
+
+def distribuer_valeur(
+    coffre: dict, secret: dict, data_dir: str | Path | None = None
+) -> str:
+    """Déchiffre la version courante d'un secret (après contrôles d'octroi
+    faits par le routeur). Refuse un secret expiré (410 côté routeur)."""
+    base = resolve_data_dir(data_dir)
+    if est_expire(secret.get("expiration")):
+        raise SecretExpireError("Secret expiré")
+    dek = ensure_dek(coffre, base)
+    with _lock:
+        _, valeurs = _load_tout(base)
+        versions = valeurs["versions"].get(secret["id"], [])
+        courante = None
+        for version in versions:
+            if version["numero"] == secret["version_courante"]:
+                courante = version
+                break
+        if courante is None:
+            raise IntrouvableError("Version courante introuvable")
+        return decrypt_secret_value(
+            dek,
+            secret["id"],
+            secret["version_courante"],
+            courante["nonce"],
+            courante["ciphertext"],
+        )
+
+
+def toucher_utilisation(secret_id: str, data_dir: str | Path | None = None) -> None:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        data, _ = _load_tout(base)
+        for secret in data["secrets"]:
+            if secret["id"] == secret_id:
+                secret["derniere_utilisation"] = _now_iso()
+                _save_coffres(base, data)
+                return
+
+
+def rotate_dek(coffre_slug: str, data_dir: str | Path | None = None) -> dict:
+    """Rotation DEK : re-chiffre toutes les versions du coffre avec une
+    nouvelle DEK, puis remplace la DEK chiffrée du coffre."""
+    base = resolve_data_dir(data_dir)
+    coffre = get_coffre_or_404(coffre_slug, base)
+    kek = load_kek()
+    ancienne_dek = ensure_dek(coffre, base)
+    nouvelle_dek = generate_dek()
+    with _lock:
+        data, valeurs = _load_tout(base)
+        secrets_coffre = [s for s in data["secrets"] if s["coffre_id"] == coffre["id"]]
+        for secret in secrets_coffre:
+            nouvelles_versions = []
+            for version in valeurs["versions"].get(secret["id"], []):
+                nouvelles_versions.append(
+                    {
+                        **reencrypt_secret_value(
+                            ancienne_dek,
+                            nouvelle_dek,
+                            secret["id"],
+                            version["numero"],
+                            version["nonce"],
+                            version["ciphertext"],
+                        ),
+                        "numero": version["numero"],
+                        "cree_le": version["cree_le"],
+                        "remplacee_le": version["remplacee_le"],
+                    }
+                )
+            valeurs["versions"][secret["id"]] = nouvelles_versions
+        for item in data["coffres"]:
+            if item["id"] == coffre["id"]:
+                item["dek_chiffree"] = wrap_dek(nouvelle_dek, kek, item["id"])
+                coffre = item
+                break
+        _save_coffres(base, data)
+        _save_valeurs(base, valeurs)
+        return coffre
+
+
+def rotate_kek(nouvelle_kek_b64: str, data_dir: str | Path | None = None) -> int:
+    """Rotation KEK : re-chiffre uniquement les DEK avec la nouvelle KEK.
+    L'opérateur doit ensuite remplacer OBSIGATE_KEK / OBSIGATE_KEK_FILE
+    par la nouvelle valeur (cette fonction ne touche pas à l'environnement).
+    Retourne le nombre de coffres re-chiffrés."""
+    base = resolve_data_dir(data_dir)
+    ancienne_kek = load_kek()
+    nouvelle_kek = parse_kek(nouvelle_kek_b64)
+    count = 0
+    with _lock:
+        data = _load_coffres(base)
+        for coffre in data["coffres"]:
+            if not coffre.get("dek_chiffree"):
+                continue
+            dek = unwrap_dek(coffre["dek_chiffree"], ancienne_kek, coffre["id"])
+            coffre["dek_chiffree"] = wrap_dek(dek, nouvelle_kek, coffre["id"])
+            count += 1
+        _save_coffres(base, data)
+        return count
+
+
+# ---------------------------------------------------------------------------
+# Consommateurs (jetons serveur-à-serveur, hachés, révocables par jti)
+# ---------------------------------------------------------------------------
+
+def _hash_jeton(jeton: str) -> str:
+    return hashlib.sha256(jeton.encode("utf-8")).hexdigest()
+
+
+def list_consommateurs(data_dir: str | Path | None = None) -> list:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        return [
+            {k: v for k, v in c.items() if k != "jeton_hash"}
+            for c in _load_coffres(base)["consommateurs"]
+        ]
+
+
+def create_consommateur(
+    nom: str, cree_par: str, data_dir: str | Path | None = None
+) -> tuple[dict, str]:
+    """Crée un consommateur et retourne (métadonnées, jeton à usage unique).
+    Le jeton n'est JAMAIS persisté : seul son hash SHA-256 est stocké."""
+    base = resolve_data_dir(data_dir)
+    nom = (nom or "").strip()
+    if not nom or len(nom) > 128:
+        raise ValidationError("Nom de consommateur requis (128 caractères max)")
+    jeton = "csk_" + _secrets.token_urlsafe(32)
+    with _lock:
+        data = _load_coffres(base)
+        consommateur = {
+            "id": str(uuid.uuid4()),
+            "nom": nom,
+            "jeton_hash": _hash_jeton(jeton),
+            "jeton_prefixe": jeton[:12],
+            "statut": "actif",
+            "cree_par": cree_par,
+            "cree_le": _now_iso(),
+        }
+        data["consommateurs"].append(consommateur)
+        _save_coffres(base, data)
+        public = {k: v for k, v in consommateur.items() if k != "jeton_hash"}
+        return public, jeton
+
+
+def verifier_jeton_consommateur(
+    jeton: str, data_dir: str | Path | None = None
+) -> dict | None:
+    """Retourne le consommateur (sans hash) si le jeton correspond, sinon None."""
+    if not jeton:
+        return None
+    base = resolve_data_dir(data_dir)
+    cible = _hash_jeton(jeton)
+    with _lock:
+        for consommateur in _load_coffres(base)["consommateurs"]:
+            if hmac.compare_digest(consommateur.get("jeton_hash", ""), cible):
+                return {
+                    k: v for k, v in consommateur.items() if k != "jeton_hash"
+                }
+    return None
+
+
+def revoquer_consommateur(
+    consommateur_id: str, data_dir: str | Path | None = None
+) -> dict:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        data = _load_coffres(base)
+        for consommateur in data["consommateurs"]:
+            if consommateur["id"] == consommateur_id:
+                consommateur["statut"] = "revoque"
+                _save_coffres(base, data)
+                return {
+                    k: v
+                    for k, v in consommateur.items()
+                    if k != "jeton_hash"
+                }
+    raise IntrouvableError("Consommateur introuvable")
+
+
+# ---------------------------------------------------------------------------
+# Octrois
+# ---------------------------------------------------------------------------
+
+def list_octrois(coffre_id: str, data_dir: str | Path | None = None) -> list:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        return [
+            o for o in _load_coffres(base)["octrois"] if o["coffre_id"] == coffre_id
+        ]
+
+
+def compter_octrois_actifs(
+    secret_id: str, data_dir: str | Path | None = None
+) -> int:
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        return sum(
+            1
+            for o in _load_coffres(base)["octrois"]
+            if o.get("secret_id") == secret_id
+            and o["statut"] == "actif"
+            and not est_expire(o.get("expiration"))
+        )
+
+
+def create_octroi(
+    coffre: dict,
+    cible_type: str,
+    cible_id: str,
+    secret_id: str | None,
+    cree_par: str,
+    expiration: str | None = None,
+    data_dir: str | Path | None = None,
+) -> dict:
+    base = resolve_data_dir(data_dir)
+    if cible_type not in CIBLES_OCTROI:
+        raise ValidationError(f"Type de cible invalide : {cible_type}")
+    if not (cible_id or "").strip():
+        raise ValidationError("Cible d'octroi requise")
+    if secret_id is not None:
+        with _lock:
+            data = _load_coffres(base)
+            if not any(
+                s["id"] == secret_id and s["coffre_id"] == coffre["id"]
+                for s in data["secrets"]
+            ):
+                raise IntrouvableError("Secret introuvable dans ce coffre")
+    if expiration is None:
+        expiration = (datetime.now(timezone.utc) + timedelta(days=DUREE_OCTROI_DEFAUT_JOURS)).isoformat()
+    else:
+        _parse_date(expiration)
+    with _lock:
+        data = _load_coffres(base)
+        octroi = {
+            "id": str(uuid.uuid4()),
+            "coffre_id": coffre["id"],
+            "secret_id": secret_id,
+            "cible_type": cible_type,
+            "cible_id": cible_id.strip(),
+            "expiration": expiration,
+            "statut": "actif",
+            "cree_par": cree_par,
+            "cree_le": _now_iso(),
+            "revoque_le": None,
+        }
+        data["octrois"].append(octroi)
+        _save_coffres(base, data)
+        return octroi
+
+
+def revoquer_octroi(octroi_id: str, data_dir: str | Path | None = None) -> dict:
+    """Révocation : statut -> revoque (conservé pour distinguer
+    octroi_revoque de octroi_absent à la distribution)."""
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        data = _load_coffres(base)
+        for octroi in data["octrois"]:
+            if octroi["id"] == octroi_id:
+                octroi["statut"] = "revoque"
+                octroi["revoque_le"] = _now_iso()
+                _save_coffres(base, data)
+                return octroi
+    raise IntrouvableError("Octroi introuvable")
+
+
+def statut_octroi_consommateur(
+    consommateur_id: str,
+    coffre_id: str,
+    secret_id: str,
+    data_dir: str | Path | None = None,
+) -> tuple[str, dict | None]:
+    """Statut d'accès d'un consommateur à un secret.
+
+    Retourne ("ok" | "revoque" | "expire" | "absent", octroi). Un octroi
+    coffre-wide (secret_id None) couvre tous les secrets du coffre.
+    Priorité : ok > revoque > expire > absent (un octroi révoqué ne doit
+    jamais être masqué par l'absence d'un autre octroi)."""
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        candidats = [
+            o
+            for o in _load_coffres(base)["octrois"]
+            if o["cible_type"] == "consommateur"
+            and o["cible_id"] == consommateur_id
+            and o["coffre_id"] == coffre_id
+            and (o.get("secret_id") is None or o["secret_id"] == secret_id)
+        ]
+    for octroi in candidats:
+        if octroi["statut"] == "actif" and not est_expire(octroi.get("expiration")):
+            return "ok", octroi
+    for octroi in candidats:
+        if octroi["statut"] == "revoque":
+            return "revoque", octroi
+    for octroi in candidats:
+        if octroi["statut"] == "actif" and est_expire(octroi.get("expiration")):
+            return "expire", octroi
+    return "absent", None
+
+
+def secrets_accessibles_utilisateur(
+    username: str, coffre: dict, data_dir: str | Path | None = None
+) -> set | None:
+    """Secrets d'un coffre visibles par un utilisateur via octroi.
+
+    Retourne None si un octroi coffre-wide actif existe (tout le coffre),
+    sinon l'ensemble des ids de secrets octroyés, sinon l'ensemble vide."""
+    base = resolve_data_dir(data_dir)
+    with _lock:
+        octrois = [
+            o
+            for o in _load_coffres(base)["octrois"]
+            if o["cible_type"] == "utilisateur"
+            and o["cible_id"] == username
+            and o["coffre_id"] == coffre["id"]
+            and o["statut"] == "actif"
+            and not est_expire(o.get("expiration"))
+        ]
+    if any(o.get("secret_id") is None for o in octrois):
+        return None
+    return {o["secret_id"] for o in octrois if o.get("secret_id")}