861 lines
29 KiB
Python
861 lines
29 KiB
Python
"""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")}
|