Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
06c12d0383 | ||
|
|
69f3968d4a | ||
|
|
d5ae66466a | ||
|
|
2dc8be3127 | ||
|
|
040e80a36d | ||
|
|
bd6b79fb41 | ||
|
|
581a31161c | ||
|
|
5f8f93ba6c | ||
|
|
40b3e9c5dd | ||
|
|
2ffc219f47 | ||
|
|
02394d3278 | ||
|
|
ff1ccca3cd | ||
|
|
fddbaec82e |
+18
-1
@@ -6,7 +6,7 @@ Format basé sur [Keep a Changelog](https://keepachangelog.com/fr/1.1.0/),
|
||||
et [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
> **En cours de développement** : les changements à venir sont listés dans la section
|
||||
> [Unreleased](#unreleased). La dernière version livrée est **2.78.2**.
|
||||
> [Unreleased](#unreleased). La dernière version livrée est **2.79.0**.
|
||||
|
||||
---
|
||||
|
||||
@@ -14,6 +14,23 @@ et [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
---
|
||||
|
||||
## [2.79.0] — 2026-10-26
|
||||
|
||||
### Ajouté
|
||||
|
||||
- **Panneau Spark : version du service d'agents visible en mode
|
||||
sidebar (#225)** — le pied du panneau Spark
|
||||
(`frontend/draco-panneau.js`, asset partagé des 4 applications)
|
||||
affiche la ligne discrète « Service d'agents vX », comme la
|
||||
console pleine page. La version est lue en direct dans la santé
|
||||
du service (`GET /healthz` de l'origine injectée par l'hôte), à
|
||||
l'ouverture du panneau, avec cache de session ; jamais une
|
||||
valeur codée en dur — service injoignable, rien d'affiché.
|
||||
Demande de Bruno (2026-10-26), carte fusion #64 ; même
|
||||
livraison : TikTak 0.61.3, Flowdeck 7.83.1, Kanbante 0.37.1.
|
||||
|
||||
---
|
||||
|
||||
## [2.78.2] — 2026-10-26
|
||||
|
||||
### Corrigé
|
||||
|
||||
+3
-3
@@ -4,7 +4,7 @@
|
||||
|
||||
**Porte d'entrée web ultra-léger pour vos vaults Obsidian** — Accédez, naviguez et recherchez dans toutes vos notes Obsidian depuis n'importe quel appareil via une interface web moderne et responsive.
|
||||
|
||||
[]()
|
||||
[]()
|
||||
[](https://opensource.org/licenses/MIT)
|
||||
[](https://www.docker.com/)
|
||||
[](https://www.python.org/)
|
||||
@@ -977,8 +977,8 @@ Ce projet est sous licence **MIT** — voir le fichier [LICENSE](LICENSE) pour l
|
||||
|
||||
## 📝 Changelog
|
||||
|
||||
Consultez le [CHANGELOG.md](./CHANGELOG.md) pour l'historique complet de toutes les versions (v1.0.0 → v2.78.2).
|
||||
Consultez le [CHANGELOG.md](./CHANGELOG.md) pour l'historique complet de toutes les versions (v1.0.0 → v2.79.0).
|
||||
|
||||
---
|
||||
|
||||
*Projet : ObsiGate | Version : 2.78.2 | Dernière mise à jour : Septembre 2026*
|
||||
*Projet : ObsiGate | Version : 2.79.0 | Dernière mise à jour : Septembre 2026*
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
**Ultra-light web gateway for your Obsidian vaults** — Access, browse, and search all your Obsidian notes from any device via a modern, responsive web interface.
|
||||
|
||||
[]()
|
||||
[]()
|
||||
[](https://opensource.org/licenses/MIT)
|
||||
[](https://www.docker.com/)
|
||||
[](https://www.python.org/)
|
||||
@@ -1152,8 +1152,8 @@ This project is licensed under the **MIT License** - see the [LICENSE](LICENSE)
|
||||
|
||||
## 📝 Changelog
|
||||
|
||||
See [CHANGELOG.md](./CHANGELOG.md) for the complete version history (v1.0.0 → v2.78.2).
|
||||
See [CHANGELOG.md](./CHANGELOG.md) for the complete version history (v1.0.0 → v2.79.0).
|
||||
|
||||
---
|
||||
|
||||
*Project: ObsiGate | Version: 2.78.2 | Last updated: September 2026*
|
||||
*Project: ObsiGate | Version: 2.79.0 | Last updated: September 2026*
|
||||
|
||||
@@ -119,6 +119,41 @@ def log_vault_remove(username: str, vault_name: str, ip: str | None = None):
|
||||
})
|
||||
|
||||
|
||||
def log_coffre_event(
|
||||
username: str,
|
||||
action: str,
|
||||
coffre: str | None = None,
|
||||
secret_ref: str | None = None,
|
||||
consommateur: str | None = None,
|
||||
resultat: str | None = None,
|
||||
contexte_id: str | None = None,
|
||||
ip: str | None = None,
|
||||
):
|
||||
"""Journal d'un événement du coffre Secrets (Phase 1b).
|
||||
|
||||
Règle absolue : aucun champ ne contient jamais une valeur de secret,
|
||||
un texte chiffré, un nonce ou un jeton. Seules des références
|
||||
(coffre:<slug>/<nom>), des identifiants et des codes de résultat.
|
||||
"""
|
||||
entry = {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"action": action,
|
||||
"username": username,
|
||||
"ip": ip or "unknown",
|
||||
}
|
||||
if coffre is not None:
|
||||
entry["coffre"] = coffre
|
||||
if secret_ref is not None:
|
||||
entry["secret_ref"] = secret_ref
|
||||
if consommateur is not None:
|
||||
entry["consommateur"] = consommateur
|
||||
if resultat is not None:
|
||||
entry["resultat"] = resultat
|
||||
if contexte_id is not None:
|
||||
entry["contexte_id"] = contexte_id
|
||||
_write_entry(entry)
|
||||
|
||||
|
||||
def get_recent_entries(limit: int = 100, action: str | None = None) -> list:
|
||||
"""Read the most recent audit log entries.
|
||||
|
||||
|
||||
@@ -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)
|
||||
@@ -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")}
|
||||
@@ -530,6 +530,7 @@ from backend.auth.middleware import (
|
||||
from backend.auth.router import router as auth_router
|
||||
from backend.bookslm_routes import router as bookslm_router
|
||||
from backend.routers.backups import router as backups_router
|
||||
from backend.routers.coffres import router as coffres_router
|
||||
from backend.routers.config import _load_config
|
||||
from backend.routers.config import router as config_router
|
||||
from backend.routers.conflicts import router as conflicts_router
|
||||
@@ -564,6 +565,7 @@ app.include_router(search_router) # ROADMAP #85 T5 — Search
|
||||
app.include_router(backups_router) # ROADMAP #85 T4 — Backups
|
||||
app.include_router(conflicts_router) # ROADMAP #85 T8 — Conflicts
|
||||
app.include_router(config_router) # ROADMAP #85 T7 — Config
|
||||
app.include_router(coffres_router) # Coffre Secrets Phase 1b — JAMAIS dans vaults.py
|
||||
app.include_router(file_chat_router) # ROADMAP #169 — Chat par fichier
|
||||
app.include_router(files_read_router) # ROADMAP #85 T6a — Files read
|
||||
app.include_router(files_media_router) # ROADMAP #85 T6c — Media/export
|
||||
|
||||
@@ -73,3 +73,5 @@ filelock>=4.2.1
|
||||
# aucune dépendance directe d'ObsiGate) — CVE-2026-114603, corrigée
|
||||
# en 1.45.1.
|
||||
opentelemetry-api>=1.45.1
|
||||
# Coffre Secrets (Phase 1a) : chiffrement par enveloppe AES-256-GCM.
|
||||
cryptography>=44.0
|
||||
|
||||
@@ -0,0 +1,606 @@
|
||||
"""API du coffre Secrets ObsiGate (Phase 1b, spec §7, §10, §11).
|
||||
|
||||
Monté dans ``backend/main.py`` à côté de ``config_router`` — JAMAIS dans
|
||||
``backend/routers/vaults.py`` : un coffre n'est pas un vault de notes, il
|
||||
ne passe ni par l'indexeur, ni par la recherche, ni par le graphe, et
|
||||
aucun outil MCP de secret n'existe.
|
||||
|
||||
Règles d'or, appliquées côté serveur à chaque requête :
|
||||
- Toutes les lectures renvoient des métadonnées seules (``has_value``,
|
||||
version, dates, octrois actifs) : jamais de valeur, jamais de texte
|
||||
chiffré, jamais de nonce.
|
||||
- Seul ``POST /api/coffres/distribuer`` rend une valeur, et uniquement à
|
||||
un consommateur authentifié par jeton (hash + jti, jamais la session
|
||||
utilisateur), avec un octroi actif et un contexte obligatoire.
|
||||
- Autorisations (spec §6) : le propriétaire seul sur son coffre
|
||||
personnel ; l'admin gère les coffres d'instance, les consommateurs et
|
||||
la KEK, jamais le coffre personnel d'autrui ; l'ACL « * » des vaults
|
||||
ne donne AUCUN droit sur un coffre (``check_vault_access`` n'est pas
|
||||
utilisé ici, volontairement).
|
||||
- Feature flag ``OBSIGATE_SECRETS_ENABLED`` (défaut désactivé) : tant
|
||||
qu'il n'est pas activé, toutes les routes répondent 503.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Request
|
||||
|
||||
from backend.audit import get_recent_entries, log_coffre_event
|
||||
from backend.auth.middleware import require_admin, require_auth
|
||||
from backend.coffre_crypto import (
|
||||
CryptoIndisponibleError,
|
||||
KEKAbsenteError,
|
||||
generate_kek_b64,
|
||||
)
|
||||
from backend.coffre_store import (
|
||||
CoffreExisteError,
|
||||
IntrouvableError,
|
||||
SecretExisteError,
|
||||
SecretExpireError,
|
||||
ValidationError,
|
||||
compter_octrois_actifs,
|
||||
compter_secrets,
|
||||
create_coffre,
|
||||
create_consommateur,
|
||||
create_octroi,
|
||||
create_secret,
|
||||
delete_secret,
|
||||
distribuer_valeur,
|
||||
get_coffre,
|
||||
get_coffre_by_id,
|
||||
get_coffre_or_404,
|
||||
get_octroi,
|
||||
get_secret,
|
||||
list_coffres,
|
||||
list_consommateurs,
|
||||
list_octrois,
|
||||
list_secrets,
|
||||
revoquer_consommateur,
|
||||
revoquer_octroi,
|
||||
rotate_dek,
|
||||
rotate_kek,
|
||||
rotate_secret,
|
||||
secret_metadata,
|
||||
secrets_accessibles_utilisateur,
|
||||
statut_octroi_consommateur,
|
||||
toucher_utilisation,
|
||||
verifier_jeton_consommateur,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("obsigate.coffres")
|
||||
|
||||
router = APIRouter(tags=["Coffres Secrets"])
|
||||
|
||||
CODES_ERREUR = {
|
||||
"octroi_absent",
|
||||
"octroi_revoque",
|
||||
"octroi_expire",
|
||||
"secret_expire",
|
||||
"kek_absente",
|
||||
"contexte_manquant",
|
||||
"consommateur_invalide",
|
||||
"consommateur_revoque",
|
||||
"crypto_indisponible",
|
||||
"secrets_desactives",
|
||||
}
|
||||
|
||||
|
||||
def secrets_actives() -> bool:
|
||||
"""Feature flag : défaut désactivé, activation explicite par l'opérateur."""
|
||||
return os.environ.get("OBSIGATE_SECRETS_ENABLED", "false").lower() in (
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
"on",
|
||||
)
|
||||
|
||||
|
||||
def _exiger_actives() -> None:
|
||||
if not secrets_actives():
|
||||
raise HTTPException(status_code=503, detail="secrets_desactives")
|
||||
|
||||
|
||||
def _username(user: dict) -> str:
|
||||
return str(user.get("username", ""))
|
||||
|
||||
|
||||
def _est_admin(user: dict) -> bool:
|
||||
return user.get("role") == "admin"
|
||||
|
||||
|
||||
def _peut_voir_coffre(user: dict, coffre: dict) -> bool:
|
||||
"""Voir les métadonnées d'un coffre."""
|
||||
if coffre["proprietaire"] == _username(user):
|
||||
return True
|
||||
if _est_admin(user) and coffre["perimetre"] == "instance":
|
||||
return True
|
||||
# Octroi utilisateur (coffre-wide ou par secret) : accès en lecture.
|
||||
accessibles = secrets_accessibles_utilisateur(_username(user), coffre)
|
||||
if accessibles is None:
|
||||
return True
|
||||
return len(accessibles) > 0
|
||||
|
||||
|
||||
def _peut_gerer_coffre(user: dict, coffre: dict) -> bool:
|
||||
"""Créer / roter / supprimer des secrets et gérer les octrois."""
|
||||
if coffre["proprietaire"] == _username(user):
|
||||
return True
|
||||
return _est_admin(user) and coffre["perimetre"] == "instance"
|
||||
|
||||
|
||||
def _exiger_voir(user: dict, coffre: dict) -> None:
|
||||
if not _peut_voir_coffre(user, coffre):
|
||||
raise HTTPException(status_code=403, detail="Accès refusé à ce coffre")
|
||||
|
||||
|
||||
def _exiger_gerer(user: dict, coffre: dict) -> None:
|
||||
if not _peut_gerer_coffre(user, coffre):
|
||||
raise HTTPException(status_code=403, detail="Accès refusé à ce coffre")
|
||||
|
||||
|
||||
def _coffre_or_404(slug: str) -> dict:
|
||||
try:
|
||||
return get_coffre_or_404(slug)
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
|
||||
def _secret_or_404(coffre: dict, nom: str) -> dict:
|
||||
try:
|
||||
secret = get_secret(coffre["id"], nom)
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
if secret is None:
|
||||
raise HTTPException(status_code=404, detail=f"Secret introuvable : {nom}")
|
||||
return secret
|
||||
|
||||
|
||||
def _metadata_coffre(coffre: dict) -> dict:
|
||||
return {
|
||||
"slug": coffre["slug"],
|
||||
"nom": coffre["nom"],
|
||||
"description": coffre.get("description", ""),
|
||||
"proprietaire": coffre["proprietaire"],
|
||||
"perimetre": coffre["perimetre"],
|
||||
"statut": coffre["statut"],
|
||||
"kek_configuree": coffre.get("dek_chiffree") is not None,
|
||||
"nb_secrets": compter_secrets(coffre["id"]),
|
||||
"cree_le": coffre["cree_le"],
|
||||
}
|
||||
|
||||
|
||||
def _metadata_secret(coffre: dict, secret: dict) -> dict:
|
||||
return secret_metadata(secret, compter_octrois_actifs(secret["id"]))
|
||||
|
||||
|
||||
def _secrets_visibles(user: dict, coffre: dict) -> list:
|
||||
"""Secrets visibles par l'utilisateur (métadonnées seules)."""
|
||||
if coffre["proprietaire"] == _username(user) or (
|
||||
_est_admin(user) and coffre["perimetre"] == "instance"
|
||||
):
|
||||
return [_metadata_secret(coffre, s) for s in list_secrets(coffre["id"])]
|
||||
accessibles = secrets_accessibles_utilisateur(_username(user), coffre)
|
||||
if accessibles is None:
|
||||
return [_metadata_secret(coffre, s) for s in list_secrets(coffre["id"])]
|
||||
return [
|
||||
_metadata_secret(coffre, s)
|
||||
for s in list_secrets(coffre["id"])
|
||||
if s["id"] in accessibles
|
||||
]
|
||||
|
||||
|
||||
def _audit(user: str, action: str, **kwargs) -> None:
|
||||
try:
|
||||
log_coffre_event(user, action, **kwargs)
|
||||
except Exception as exc: # l'audit ne doit jamais casser l'opération
|
||||
logger.warning("Échec audit coffre (%s) : %s", action, exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Coffres
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/api/coffres")
|
||||
async def api_list_coffres(current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
visibles = [c for c in list_coffres() if _peut_voir_coffre(current_user, c)]
|
||||
return [_metadata_coffre(c) for c in visibles]
|
||||
|
||||
|
||||
@router.post("/api/coffres")
|
||||
async def api_create_coffre(body: dict = Body(...), current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
slug = body.get("slug", "")
|
||||
perimetre = body.get("perimetre", "personnel")
|
||||
if perimetre == "instance" and not _est_admin(current_user):
|
||||
raise HTTPException(
|
||||
status_code=403, detail="Seul un admin peut créer un coffre d'instance"
|
||||
)
|
||||
try:
|
||||
coffre = create_coffre(
|
||||
slug=slug,
|
||||
nom=body.get("nom", ""),
|
||||
proprietaire=_username(current_user),
|
||||
perimetre=perimetre,
|
||||
description=body.get("description", ""),
|
||||
)
|
||||
except CoffreExisteError as exc:
|
||||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"coffre_cree",
|
||||
coffre=coffre["slug"],
|
||||
resultat="ok",
|
||||
)
|
||||
return _metadata_coffre(coffre)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Consommateurs (admin seulement, jeton rendu une seule fois)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/api/coffres/consommateurs")
|
||||
async def api_list_consommateurs(current_user=Depends(require_admin)):
|
||||
_exiger_actives()
|
||||
return list_consommateurs()
|
||||
|
||||
|
||||
@router.post("/api/coffres/consommateurs")
|
||||
async def api_create_consommateur(
|
||||
body: dict = Body(...), current_user=Depends(require_admin)
|
||||
):
|
||||
_exiger_actives()
|
||||
try:
|
||||
consommateur, jeton = create_consommateur(
|
||||
nom=body.get("nom", ""), cree_par=_username(current_user)
|
||||
)
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"consommateur_cree",
|
||||
consommateur=consommateur["nom"],
|
||||
resultat="ok",
|
||||
)
|
||||
# Le jeton n'est rendu qu'ici, jamais stocké en clair, jamais relisté.
|
||||
return {**consommateur, "jeton": jeton}
|
||||
|
||||
|
||||
@router.delete("/api/coffres/consommateurs/{consommateur_id}")
|
||||
async def api_revoke_consommateur(
|
||||
consommateur_id: str, current_user=Depends(require_admin)
|
||||
):
|
||||
_exiger_actives()
|
||||
try:
|
||||
consommateur = revoquer_consommateur(consommateur_id)
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"consommateur_revoque",
|
||||
consommateur=consommateur["nom"],
|
||||
resultat="ok",
|
||||
)
|
||||
return consommateur
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Distribution (serveur-à-serveur, hors session utilisateur, hors LLM)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.post("/api/coffres/distribuer")
|
||||
async def api_distribuer(request: Request, body: dict = Body(...)):
|
||||
_exiger_actives()
|
||||
reference = (body.get("reference") or "").strip()
|
||||
contexte_type = (body.get("contexte_type") or "").strip()
|
||||
contexte_id = (body.get("contexte_id") or "").strip()
|
||||
if not contexte_type or not contexte_id:
|
||||
raise HTTPException(status_code=400, detail="contexte_manquant")
|
||||
if not reference.startswith("coffre:") or "/" not in reference[7:]:
|
||||
raise HTTPException(status_code=400, detail="reference_invalide")
|
||||
slug, nom = reference[7:].split("/", 1)
|
||||
# Authentification consommateur : Bearer, hash + statut, jamais un JWT
|
||||
# de session utilisateur.
|
||||
auth = request.headers.get("Authorization", "")
|
||||
jeton = auth[7:] if auth.startswith("Bearer ") else ""
|
||||
consommateur = verifier_jeton_consommateur(jeton)
|
||||
if consommateur is None:
|
||||
_audit("consommateur", "distribution_refusee", secret_ref=reference,
|
||||
resultat="consommateur_invalide", contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=401, detail="consommateur_invalide")
|
||||
if consommateur["statut"] != "actif":
|
||||
_audit("consommateur", "distribution_refusee", secret_ref=reference,
|
||||
consommateur=consommateur["nom"],
|
||||
resultat="consommateur_revoque", contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=403, detail="consommateur_revoque")
|
||||
try:
|
||||
coffre = get_coffre(slug)
|
||||
secret = get_secret(coffre["id"], nom) if coffre else None
|
||||
except ValidationError:
|
||||
raise HTTPException(status_code=400, detail="reference_invalide") from None
|
||||
if coffre is None or secret is None or coffre["statut"] != "actif":
|
||||
# Anti-énumération : un secret inexistant se comporte comme un
|
||||
# octroi absent, il ne révèle pas l'existence du coffre/secret.
|
||||
_audit("consommateur", "distribution_refusee", secret_ref=reference,
|
||||
consommateur=consommateur["nom"],
|
||||
resultat="octroi_absent", contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=403, detail="octroi_absent")
|
||||
statut, _octroi = statut_octroi_consommateur(
|
||||
consommateur["id"], coffre["id"], secret["id"]
|
||||
)
|
||||
if statut != "ok":
|
||||
code = {
|
||||
"revoque": "octroi_revoque",
|
||||
"expire": "octroi_expire",
|
||||
"absent": "octroi_absent",
|
||||
}[statut]
|
||||
_audit("consommateur", "distribution_refusee", coffre=coffre["slug"],
|
||||
secret_ref=reference, consommateur=consommateur["nom"],
|
||||
resultat=code, contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=403, detail=code)
|
||||
try:
|
||||
valeur = distribuer_valeur(coffre, secret)
|
||||
except SecretExpireError:
|
||||
_audit("consommateur", "distribution_refusee", coffre=coffre["slug"],
|
||||
secret_ref=reference, consommateur=consommateur["nom"],
|
||||
resultat="secret_expire", contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=410, detail="secret_expire") from None
|
||||
except KEKAbsenteError:
|
||||
_audit("consommateur", "distribution_refusee", coffre=coffre["slug"],
|
||||
secret_ref=reference, consommateur=consommateur["nom"],
|
||||
resultat="kek_absente", contexte_id=contexte_id)
|
||||
raise HTTPException(status_code=503, detail="kek_absente") from None
|
||||
except CryptoIndisponibleError:
|
||||
raise HTTPException(status_code=503, detail="crypto_indisponible") from None
|
||||
toucher_utilisation(secret["id"])
|
||||
_audit("consommateur", "distribution_ok", coffre=coffre["slug"],
|
||||
secret_ref=reference, consommateur=consommateur["nom"],
|
||||
resultat="ok", contexte_id=contexte_id)
|
||||
return {
|
||||
"reference": reference,
|
||||
"valeur": valeur,
|
||||
"version": secret["version_courante"],
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Un coffre, ses secrets, ses octrois, son journal
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.get("/api/coffres/{slug}")
|
||||
async def api_get_coffre(slug: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_voir(current_user, coffre)
|
||||
return _metadata_coffre(coffre)
|
||||
|
||||
|
||||
@router.get("/api/coffres/{slug}/secrets")
|
||||
async def api_list_secrets(slug: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_voir(current_user, coffre)
|
||||
return _secrets_visibles(current_user, coffre)
|
||||
|
||||
|
||||
@router.post("/api/coffres/{slug}/secrets")
|
||||
async def api_create_secret(
|
||||
slug: str, body: dict = Body(...), current_user=Depends(require_auth)
|
||||
):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_gerer(current_user, coffre)
|
||||
try:
|
||||
secret = create_secret(
|
||||
coffre=coffre,
|
||||
nom=body.get("nom", ""),
|
||||
type_secret=body.get("type", "autre"),
|
||||
valeur=body.get("valeur", ""),
|
||||
service=body.get("service", ""),
|
||||
expiration=body.get("expiration"),
|
||||
note=body.get("note", ""),
|
||||
)
|
||||
except SecretExisteError as exc:
|
||||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except KEKAbsenteError:
|
||||
raise HTTPException(status_code=503, detail="kek_absente") from None
|
||||
except CryptoIndisponibleError:
|
||||
raise HTTPException(status_code=503, detail="crypto_indisponible") from None
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"secret_cree",
|
||||
coffre=coffre["slug"],
|
||||
secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}",
|
||||
resultat="ok",
|
||||
)
|
||||
return _metadata_secret(coffre, secret)
|
||||
|
||||
|
||||
@router.post("/api/coffres/{slug}/secrets/{nom}/rotation")
|
||||
async def api_rotate_secret(
|
||||
slug: str, nom: str, body: dict = Body(...), current_user=Depends(require_auth)
|
||||
):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_gerer(current_user, coffre)
|
||||
try:
|
||||
secret = rotate_secret(coffre, nom, body.get("valeur", ""))
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except KEKAbsenteError:
|
||||
raise HTTPException(status_code=503, detail="kek_absente") from None
|
||||
except CryptoIndisponibleError:
|
||||
raise HTTPException(status_code=503, detail="crypto_indisponible") from None
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"secret_rotation",
|
||||
coffre=coffre["slug"],
|
||||
secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}",
|
||||
resultat="ok",
|
||||
)
|
||||
return _metadata_secret(coffre, secret)
|
||||
|
||||
|
||||
@router.delete("/api/coffres/{slug}/secrets/{nom}")
|
||||
async def api_delete_secret(
|
||||
slug: str, nom: str, current_user=Depends(require_auth)
|
||||
):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_gerer(current_user, coffre)
|
||||
try:
|
||||
secret, _octrois = delete_secret(coffre, nom)
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"secret_supprime",
|
||||
coffre=coffre["slug"],
|
||||
secret_ref=f"coffre:{coffre['slug']}/{secret['nom']}",
|
||||
resultat="ok",
|
||||
)
|
||||
return {"status": "deleted", "nom": secret["nom"]}
|
||||
|
||||
|
||||
@router.get("/api/coffres/{slug}/octrois")
|
||||
async def api_list_octrois(slug: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_voir(current_user, coffre)
|
||||
return list_octrois(coffre["id"])
|
||||
|
||||
|
||||
@router.post("/api/coffres/{slug}/octrois")
|
||||
async def api_create_octroi(
|
||||
slug: str, body: dict = Body(...), current_user=Depends(require_auth)
|
||||
):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_gerer(current_user, coffre)
|
||||
secret_id = None
|
||||
if body.get("secret_nom"):
|
||||
secret = _secret_or_404(coffre, body["secret_nom"])
|
||||
secret_id = secret["id"]
|
||||
try:
|
||||
octroi = create_octroi(
|
||||
coffre=coffre,
|
||||
cible_type=body.get("cible_type", ""),
|
||||
cible_id=body.get("cible_id", ""),
|
||||
secret_id=secret_id,
|
||||
cree_par=_username(current_user),
|
||||
expiration=body.get("expiration"),
|
||||
)
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
except ValidationError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"octroi_cree",
|
||||
coffre=coffre["slug"],
|
||||
secret_ref=(
|
||||
f"coffre:{coffre['slug']}/{body.get('secret_nom', '*')}"
|
||||
),
|
||||
resultat="ok",
|
||||
)
|
||||
return octroi
|
||||
|
||||
|
||||
@router.delete("/api/coffres/octrois/{octroi_id}")
|
||||
async def api_revoke_octroi(octroi_id: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
# Le gestionnaire du coffre concerné (ou l'admin d'instance) seul peut
|
||||
# révoquer : on résout le coffre depuis l'octroi avant toute action.
|
||||
octroi = get_octroi(octroi_id)
|
||||
if octroi is None:
|
||||
raise HTTPException(status_code=404, detail="Octroi introuvable")
|
||||
coffre = get_coffre_by_id(octroi["coffre_id"])
|
||||
if coffre is None:
|
||||
raise HTTPException(status_code=404, detail="Coffre introuvable")
|
||||
_exiger_gerer(current_user, coffre)
|
||||
try:
|
||||
octroi = revoquer_octroi(octroi_id)
|
||||
except IntrouvableError as exc:
|
||||
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
||||
_audit(
|
||||
_username(current_user),
|
||||
"octroi_revoque",
|
||||
coffre=coffre["slug"],
|
||||
resultat="ok",
|
||||
)
|
||||
return octroi
|
||||
|
||||
|
||||
@router.get("/api/coffres/{slug}/journal")
|
||||
async def api_journal(slug: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_voir(current_user, coffre)
|
||||
# Défense en profondeur : même si une entrée contenait un jour un champ
|
||||
# sensible, il ne sort jamais par cette route.
|
||||
nettoyees = []
|
||||
for entry in get_recent_entries(limit=100):
|
||||
if entry.get("coffre") != slug:
|
||||
continue
|
||||
nettoyees.append(
|
||||
{
|
||||
k: v
|
||||
for k, v in entry.items()
|
||||
if k not in ("valeur", "ciphertext", "nonce", "jeton", "dek_chiffree")
|
||||
}
|
||||
)
|
||||
return nettoyees
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Administration KEK / rotations (admin seulement)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@router.post("/api/coffres/admin/generer-kek")
|
||||
async def api_generer_kek(current_user=Depends(require_admin)):
|
||||
_exiger_actives()
|
||||
# La KEK générée n'est rendue qu'une fois, ici, et jamais stockée par
|
||||
# ObsiGate : l'opérateur doit la sauvegarder hors dépôt immédiatement.
|
||||
return {"kek": generate_kek_b64()}
|
||||
|
||||
|
||||
@router.post("/api/coffres/{slug}/rotation-dek")
|
||||
async def api_rotate_dek(slug: str, current_user=Depends(require_auth)):
|
||||
_exiger_actives()
|
||||
coffre = _coffre_or_404(slug)
|
||||
_exiger_gerer(current_user, coffre)
|
||||
try:
|
||||
coffre = rotate_dek(slug)
|
||||
except KEKAbsenteError:
|
||||
raise HTTPException(status_code=503, detail="kek_absente") from None
|
||||
except CryptoIndisponibleError:
|
||||
raise HTTPException(status_code=503, detail="crypto_indisponible") from None
|
||||
_audit(_username(current_user), "dek_rotation", coffre=coffre["slug"],
|
||||
resultat="ok")
|
||||
return _metadata_coffre(coffre)
|
||||
|
||||
|
||||
@router.post("/api/coffres/admin/rotation-kek")
|
||||
async def api_rotate_kek(
|
||||
body: dict = Body(...), current_user=Depends(require_admin)
|
||||
):
|
||||
_exiger_actives()
|
||||
try:
|
||||
count = rotate_kek(body.get("nouvelle_kek", ""))
|
||||
except KEKAbsenteError:
|
||||
raise HTTPException(status_code=503, detail="kek_absente") from None
|
||||
except Exception as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
_audit(_username(current_user), "kek_rotation", resultat="ok")
|
||||
return {"status": "ok", "coffres_rechiffres": count}
|
||||
Generated
+1
-1
@@ -2626,7 +2626,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "obsigate-desktop"
|
||||
version = "2.78.2"
|
||||
version = "2.79.0"
|
||||
dependencies = [
|
||||
"chrono",
|
||||
"env_logger",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "obsigate-desktop"
|
||||
version = "2.78.2"
|
||||
version = "2.79.0"
|
||||
description = "ObsiGate Desktop — Porte d'entrée native pour vos vaults Obsidian"
|
||||
authors = ["Bruno Charest"]
|
||||
edition = "2021"
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"$schema": "https://raw.githubusercontent.com/nicedoc/obsigate/main/desktop/tauri.conf.schema.json",
|
||||
"productName": "ObsiGate",
|
||||
"version": "2.78.2",
|
||||
"version": "2.79.0",
|
||||
"identifier": "com.obsigate.desktop",
|
||||
"build": {
|
||||
"frontendDist": "../frontend",
|
||||
|
||||
+20
-1
@@ -1,6 +1,6 @@
|
||||
# ObsiGate — Roadmap
|
||||
|
||||
> **Version :** 2.78.2 | **Dernière mise à jour :** 2026-10-26
|
||||
> **Version :** 2.79.0 | **Dernière mise à jour :** 2026-10-26
|
||||
> **Ce fichier ne contient que le travail à venir** (🔵 En cours + ⚪ Backlog) et un index compact
|
||||
> vers les fonctionnalités livrées.
|
||||
> - **Méthode de livraison à appliquer pour toute tâche : [DELIVERY_WORKFLOW.md](./DELIVERY_WORKFLOW.md)**
|
||||
@@ -30,6 +30,24 @@
|
||||
|
||||
## 🔵 En cours
|
||||
|
||||
### 225. Fusion — panneau Spark : version du service d'agents visible en mode sidebar
|
||||
|
||||
- **Effort :** ~0,1 jour | **Impact :** 🟢 | **Ouvert :** 2026-10-26 | **Statut :** ✅ livré le 2026-10-26 (version du hook — voir CHANGELOG ; redéploiement Docker requis)
|
||||
- **Fiche :** [features/fusion-panneau-version-agent-225.md](./features/fusion-panneau-version-agent-225.md)
|
||||
- **Description :** demande de Bruno (2026-10-26, carte fusion #64) : la
|
||||
version « Service d'agents vX », déjà visible dans la console pleine
|
||||
page, doit se lire aussi dans le **panneau natif des apps** (mode
|
||||
sidebar). Le pied du panneau Spark affiche désormais cette ligne,
|
||||
discrète et permanente, lue **en direct** dans la santé du service
|
||||
(`GET /healthz` de l'origine injectée par l'hôte) à l'ouverture du
|
||||
panneau, avec cache de session — jamais une valeur codée en dur ;
|
||||
service injoignable = rien d'affiché. Même livraison : TikTak
|
||||
0.61.3, Flowdeck 7.83.1, Kanbante 0.37.1.
|
||||
- **Sous-tâches :**
|
||||
- [x] Ligne de version dans le pied du panneau (canonique TikTak, propagée aux 4 apps)
|
||||
- [x] Test `tests/test_draco_barre.py::test_panneau_version_service_affichee_225` + épreuve comportementale jsdom sur les 4 copies
|
||||
- [x] Release publiée — CI verte à constater, redéploiement Docker (geste de Bruno)
|
||||
|
||||
### 224. Sécurité des dépendances — pip-audit vierge (brotli, filelock, opentelemetry-api, pypdf)
|
||||
|
||||
- **Effort :** ~0,1 jour | **Impact :** 🔴 | **Ouvert :** 2026-10-25 | **Statut :** ✅ livré le 2026-10-25 (version du hook — voir CHANGELOG ; redéploiement Docker requis)
|
||||
@@ -1150,6 +1168,7 @@
|
||||
| 170 | Assistant IA — tâches planifiées type cron (create/append/notify, interval/daily/once, tick 60 s, échec notifié) | Unreleased | [features/agent-phase4-166-168-170.md](./features/agent-phase4-166-168-170.md) |
|
||||
| BUG-119 | Panneau Spark — « mode console » bloqué par la CSP (`frame-src` sans l'origine du service d'agents) | 2.77.1 | [ISSUES_TODOLIST.md](./ISSUES_TODOLIST.md) |
|
||||
| BUG-120 | Menu inter-apps — les URL sœurs (`window.DRACO_URLS`) dérivent l'environnement de l'hôte de l'agent (`-dev` en dev), explicite prioritaire, Gitea inchangée (suite de #223) | 2.78.2 | [ISSUES_TODOLIST.md](./ISSUES_TODOLIST.md) |
|
||||
| 225 | Panneau Spark — version du service d'agents visible en mode sidebar (pied du panneau, lue en direct via `GET /healthz`) | 2.79.0 | [features/fusion-panneau-version-agent-225.md](./features/fusion-panneau-version-agent-225.md) |
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
# #225 — Fusion : panneau Spark, version du service d'agents visible en mode sidebar
|
||||
|
||||
**Contexte.** La version « Service d'agents vX » du service d'agents
|
||||
est visible dans la console pleine page (pilule d'en-tête, rendue par
|
||||
le service lui-même), mais pas dans le **panneau natif des
|
||||
applications** (mode sidebar) — or c'est là que Bruno vit au
|
||||
quotidien. Demande de Bruno (2026-10-26), carte fusion #64 : la
|
||||
version doit se lire dans le panneau ouvert, sans ouvrir la console.
|
||||
|
||||
**Changement.** Dans `frontend/draco-panneau.js` — asset **vendangé**,
|
||||
maintenu en canonique dans TikTak et recopié dans les 4 applications
|
||||
(TikTak, Flowdeck, Kanbante, ObsiGate ; chaque copie garde son bloc
|
||||
d'injection d'origine propre à l'hôte, le corps est commun) :
|
||||
|
||||
- le pied du panneau gagne une ligne dédiée
|
||||
`.draco-panneau-version`, discrète (11,5 px, opacité 0,6,
|
||||
monochrome, aucun emoji), **masquée tant qu'aucune version n'est
|
||||
connue** ;
|
||||
- à l'ouverture du panneau, `chargerVersion()` interroge la santé
|
||||
du service — `GET /healthz` de l'origine que le panneau connaît
|
||||
déjà (celle injectée par l'hôte, ici `window.DRACO_URLS.agent`) —
|
||||
et affiche « Service d'agents vX » depuis le champ `version` de
|
||||
la réponse (la même source que la pilule de la console) ;
|
||||
- cache de session de 5 minutes, rafraîchissement au rythme des
|
||||
ouvertures du panneau (comme la vérification d'identité) ;
|
||||
- **jamais de valeur codée en dur** : réponse illisible ou service
|
||||
injoignable = la ligne reste masquée — pas de fausse version, pas
|
||||
d'erreur visible.
|
||||
|
||||
**Preuve.** Épreuve comportementale (jsdom) exécutée sur les 4
|
||||
copies avec le contrat d'injection de chaque hôte : santé joignable
|
||||
→ la ligne affiche « Service d'agents v0.38.0 » (version lue sur
|
||||
l'origine de l'agent) ; santé injoignable → rien d'affiché ; hôte
|
||||
sans injection → aucun appel. Test de garde :
|
||||
`tests/test_draco_barre.py::test_panneau_version_service_affichee_225`.
|
||||
|
||||
**Même livraison** (carte fusion #64) : TikTak 0.61.3, Flowdeck
|
||||
7.83.1, Kanbante 0.37.1. Côté ObsiGate, la prise d'effet attend le
|
||||
**redéploiement Docker** (geste de Bruno), comme les items panneau
|
||||
précédents.
|
||||
@@ -42,6 +42,7 @@
|
||||
var etatConnu = false;
|
||||
var root = null, msgsEl = null, etatEl = null, inputEl = null, sendBtn = null;
|
||||
var titreEl = null, tiroirEl = null;
|
||||
var versionEl = null, versionConnue = "", versionMaj = 0;
|
||||
|
||||
function domaine() {
|
||||
try { return localStorage.getItem("draco_domain") || "pro"; } catch (e) { return "pro"; }
|
||||
@@ -269,6 +270,7 @@
|
||||
' <input type="text" placeholder="Écrire à Spark…" autocomplete="off" aria-label="Message à Spark">' +
|
||||
' <button type="submit">Envoyer</button></form>' +
|
||||
' <div class="draco-panneau-note">Conversation réelle, sous-agents (J3) et missions longues (J5) vers les exécuteurs configurés ; toute écriture dans les apps passe par une approbation.</div>' +
|
||||
' <div class="draco-panneau-version" style="margin-top:6px;font-size:11.5px;opacity:.6" hidden></div>' +
|
||||
'</div>';
|
||||
document.body.appendChild(root);
|
||||
msgsEl = root.querySelector(".draco-panneau-msgs");
|
||||
@@ -277,6 +279,7 @@
|
||||
sendBtn = root.querySelector("button[type=submit]");
|
||||
titreEl = root.querySelector(".d-titre");
|
||||
tiroirEl = root.querySelector(".draco-discussions");
|
||||
versionEl = root.querySelector(".draco-panneau-version");
|
||||
root.querySelector(".fermer").addEventListener("click", fermer);
|
||||
root.querySelector('[data-disc="liste"]').addEventListener("click", basculerTiroir);
|
||||
root.querySelector('[data-disc="nouvelle"]').addEventListener("click", nouvelleDiscussion);
|
||||
@@ -331,6 +334,35 @@
|
||||
titreEl.title = titreEl.textContent;
|
||||
}
|
||||
|
||||
/* ── Version du service : lue en direct dans la santé de
|
||||
* l'agent (GET /healthz → champ « version », la même source que
|
||||
* la console) à l'origine que le panneau connaît déjà. Cache de
|
||||
* session (5 min), rafraîchie à l'ouverture du panneau, au même
|
||||
* rythme que la vérification d'identité. Jamais de valeur codée
|
||||
* en dur : source injoignable ou réponse illisible = la ligne
|
||||
* reste masquée — pas de fausse version, pas d'erreur visible. */
|
||||
function afficherVersion() {
|
||||
if (!versionEl) return;
|
||||
if (versionConnue) {
|
||||
versionEl.textContent = "Service d'agents v" + versionConnue;
|
||||
versionEl.hidden = false;
|
||||
}
|
||||
}
|
||||
function chargerVersion() {
|
||||
if (!API || !versionEl) return;
|
||||
if (versionConnue && Date.now() - versionMaj < 300000) { afficherVersion(); return; }
|
||||
fetch(API + "/healthz")
|
||||
.then(function (r) { if (!r.ok) throw new Error("HTTP " + r.status); return r.json(); })
|
||||
.then(function (d) {
|
||||
if (d && typeof d.version === "string" && /^[0-9]+\.[0-9]+\.[0-9]+/.test(d.version)) {
|
||||
versionConnue = d.version;
|
||||
versionMaj = Date.now();
|
||||
afficherVersion();
|
||||
}
|
||||
})
|
||||
.catch(function () { /* service injoignable : rien d'affiché */ });
|
||||
}
|
||||
|
||||
function verifierIdentite() {
|
||||
if (etatConnu) return;
|
||||
fetch(API + "/api/v1/me", { credentials: "include" })
|
||||
@@ -807,6 +839,7 @@
|
||||
root.classList.add("open");
|
||||
appliquerTheme();
|
||||
verifierIdentite();
|
||||
chargerVersion();
|
||||
setTimeout(function () { inputEl.focus(); }, 300);
|
||||
}
|
||||
function fermer() { if (root) root.classList.remove("open"); }
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -4,6 +4,7 @@ import { api, AuthManager } from './auth.js';
|
||||
import { populateCustomDropdown, TabManager, closeMobileSidebar, ContextMenuManager } from './ui.js';
|
||||
import { _populateRecentVaultFilter, switchSidebarTab, filterAIHistory, filterRecentFiles } from './config.js';
|
||||
import { el, icon, getVaultIcon, smallBadge, attachTreeItemActionButton, attachTreeItemLongPress, showWelcome, appendHighlightedText, filterSavedSearches } from './viewer.js';
|
||||
import { openCoffresView, openCoffresNewSecret, openCoffresAcces, openCoffresJournal, getCoffresForSidebar, invalidateCoffresCache } from './coffres.js';
|
||||
import { performAdvancedSearch } from './search.js';
|
||||
import { t } from './i18n.js';
|
||||
import { filterChatMessages as _filterChat } from './filechat.js';
|
||||
@@ -46,6 +47,168 @@ function _sharedIcon(size = 16) {
|
||||
return i;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Coffre Secrets — nœud virtuel « Secrets » sous le Home (Phase 1c, #19).
|
||||
// Patron exact du dossier virtuel « Partage » (#196) : inséré à la racine
|
||||
// du Home de l'utilisateur courant seulement, sans dossier sur disque, sans
|
||||
// passer par l'indexeur. Contrairement à « Partage », le nœud est permanent
|
||||
// (même sans coffre) : cliquer dessus ouvre la vue Secrets, qui crée le
|
||||
// coffre personnel à la première ouverture si besoin.
|
||||
// AUCUNE action de fichier n'est attachée à ce nœud ni à ses enfants :
|
||||
// ni attachTreeItemActionButton, ni ContextMenuManager (qui proposerait
|
||||
// renommer/déplacer/supprimer). Le menu ⋮ est un menu propre ci-dessous.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Icône cadenas (U+1F510) du nœud Secrets, distincte de la maison et du diamant. */
|
||||
function _secretsIcon() {
|
||||
const s = document.createElement("span");
|
||||
s.className = "coffres-icon-lock";
|
||||
s.textContent = "\u{1F510}";
|
||||
return s;
|
||||
}
|
||||
|
||||
function _isOwnHome(vaultName) {
|
||||
const me = (AuthManager.getUser() || {}).username;
|
||||
return !!me && vaultName === `home-${me}`;
|
||||
}
|
||||
|
||||
/** Construit le nœud Secrets + son conteneur d'enfants (coffres). */
|
||||
async function _buildSecretsNode(vaultName) {
|
||||
const pair = document.createDocumentFragment();
|
||||
const item = el("div", { class: "tree-item", "data-vault": vaultName, "data-path": "Secrets", "data-type": "virtual-secrets" }, [icon("chevron-right", 14), _secretsIcon(), el("span", { class: "tree-item-text" }, [document.createTextNode(t("coffres.secrets"))])]);
|
||||
const sub = el("div", { class: "tree-children collapsed", id: `dir-${vaultName}-Secrets` });
|
||||
item.addEventListener("click", async () => {
|
||||
scrollTreeItemIntoView(item, false);
|
||||
if (sub.classList.contains("collapsed")) {
|
||||
if (sub.children.length === 0) await _fillSecretsNode(item, sub, vaultName);
|
||||
sub.classList.remove("collapsed");
|
||||
const chev = item.querySelector("[data-lucide]");
|
||||
if (chev) chev.setAttribute("data-lucide", "chevron-down");
|
||||
safeCreateIcons();
|
||||
} else {
|
||||
sub.classList.add("collapsed");
|
||||
const chev = item.querySelector("[data-lucide]");
|
||||
if (chev) chev.setAttribute("data-lucide", "chevron-right");
|
||||
safeCreateIcons();
|
||||
}
|
||||
await openCoffresView();
|
||||
closeMobileSidebar();
|
||||
});
|
||||
item.addEventListener("contextmenu", (e) => {
|
||||
e.preventDefault();
|
||||
_openSecretsMenu(e.clientX, e.clientY);
|
||||
});
|
||||
const actionBtn = el("button", { class: "tree-item-action-btn", type: "button", title: t("coffres.secrets") }, [document.createTextNode("\u22EE")]);
|
||||
actionBtn.addEventListener("click", (e) => {
|
||||
e.stopPropagation();
|
||||
const r = actionBtn.getBoundingClientRect();
|
||||
_openSecretsMenu(Math.min(r.right, window.innerWidth - 16), r.bottom);
|
||||
});
|
||||
item.appendChild(actionBtn);
|
||||
pair.appendChild(item);
|
||||
pair.appendChild(sub);
|
||||
// Compteur et enfants, en arrière-plan : le nœud existe même si l'API échoue.
|
||||
_fillSecretsNode(item, sub, vaultName).catch(() => {});
|
||||
return pair;
|
||||
}
|
||||
|
||||
/** Remplit le compteur et les enfants (coffres) d'un nœud Secrets existant. */
|
||||
async function _fillSecretsNode(item, sub, vaultName) {
|
||||
const data = await getCoffresForSidebar();
|
||||
item.querySelectorAll(".badge-small, .coffres-alert-dot").forEach((n) => n.remove());
|
||||
if (!data.disabled && data.personnel) {
|
||||
if (!data.personnel.kek_configuree) {
|
||||
const dot = document.createElement("span");
|
||||
dot.className = "coffres-alert-dot";
|
||||
dot.textContent = "\u26A0";
|
||||
dot.title = t("coffres.kek_absente");
|
||||
item.appendChild(dot);
|
||||
} else {
|
||||
item.appendChild(smallBadge(data.personnel.nb_secrets));
|
||||
}
|
||||
}
|
||||
sub.innerHTML = "";
|
||||
const coffres = (data.coffres || []).slice().sort((a, b) => {
|
||||
const aPers = a.proprietaire === (AuthManager.getUser() || {}).username && a.perimetre === "personnel" ? 0 : 1;
|
||||
const bPers = b.proprietaire === (AuthManager.getUser() || {}).username && b.perimetre === "personnel" ? 0 : 1;
|
||||
return aPers - bPers || String(a.nom).localeCompare(String(b.nom));
|
||||
});
|
||||
coffres.forEach((c) => {
|
||||
const me = (AuthManager.getUser() || {}).username;
|
||||
const label = c.proprietaire === me ? c.nom : `${c.nom} (${c.proprietaire})`;
|
||||
const child = el("div", { class: "tree-item", "data-vault": vaultName, "data-path": `Secrets/${c.slug}`, "data-type": "virtual-coffre", "data-coffre-slug": c.slug }, [icon("lock", 16), el("span", { class: "tree-item-text" }, [document.createTextNode(label)]), smallBadge(c.nb_secrets)]);
|
||||
child.addEventListener("click", async () => {
|
||||
scrollTreeItemIntoView(child, false);
|
||||
await openCoffresView(c.slug);
|
||||
closeMobileSidebar();
|
||||
});
|
||||
child.addEventListener("contextmenu", (e) => {
|
||||
e.preventDefault();
|
||||
_openSecretsMenu(e.clientX, e.clientY);
|
||||
});
|
||||
sub.appendChild(child);
|
||||
});
|
||||
safeCreateIcons();
|
||||
}
|
||||
|
||||
/** Rafraîchit tous les nœuds Secrets affichés (après création, rotation, révocation…). */
|
||||
async function _refreshSecretsNodes() {
|
||||
const items = Array.from(document.querySelectorAll('[data-type="virtual-secrets"]'));
|
||||
for (const item of items) {
|
||||
const vaultName = item.dataset.vault;
|
||||
const sub = item.nextElementSibling;
|
||||
if (sub && sub.classList.contains("tree-children")) {
|
||||
await _fillSecretsNode(item, sub, vaultName);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (!window.__obsigateSecretsListener) {
|
||||
window.__obsigateSecretsListener = true;
|
||||
document.addEventListener("obsigate:coffres-changed", () => {
|
||||
invalidateCoffresCache();
|
||||
_refreshSecretsNodes().catch(() => {});
|
||||
});
|
||||
}
|
||||
|
||||
/** Menu ⋮ du nœud Secrets : Ouvrir, Nouveau secret, Gérer les accès, Journal.
|
||||
* Menu propre, jamais ContextMenuManager (pas d'action de fichier). */
|
||||
function _openSecretsMenu(x, y) {
|
||||
document.querySelectorAll(".coffres-context-menu").forEach((n) => n.remove());
|
||||
const menu = el("div", { class: "coffres-context-menu" }, []);
|
||||
const entries = [
|
||||
[t("coffres.detail"), () => openCoffresView()],
|
||||
[t("coffres.new_secret"), () => openCoffresNewSecret()],
|
||||
[t("coffres.acces"), () => openCoffresAcces()],
|
||||
[t("coffres.journal"), () => openCoffresJournal()],
|
||||
];
|
||||
entries.forEach(([label, run]) => {
|
||||
const b = el("button", { class: "coffres-context-item", type: "button" }, [document.createTextNode(label)]);
|
||||
b.addEventListener("click", async () => {
|
||||
menu.remove();
|
||||
await run();
|
||||
closeMobileSidebar();
|
||||
});
|
||||
menu.appendChild(b);
|
||||
});
|
||||
document.body.appendChild(menu);
|
||||
const rect = menu.getBoundingClientRect();
|
||||
menu.style.left = Math.max(8, Math.min(x, window.innerWidth - rect.width - 8)) + "px";
|
||||
menu.style.top = Math.max(8, Math.min(y, window.innerHeight - rect.height - 8)) + "px";
|
||||
const closer = (e) => {
|
||||
if (!menu.contains(e.target)) {
|
||||
menu.remove();
|
||||
document.removeEventListener("click", closer, true);
|
||||
document.removeEventListener("keydown", onKey);
|
||||
}
|
||||
};
|
||||
const onKey = (e) => { if (e.key === "Escape") { menu.remove(); document.removeEventListener("click", closer, true); document.removeEventListener("keydown", onKey); } };
|
||||
setTimeout(() => {
|
||||
document.addEventListener("click", closer, true);
|
||||
document.addEventListener("keydown", onKey);
|
||||
}, 0);
|
||||
}
|
||||
|
||||
/** Fichier partagé (créé ou reçu) → icône share-2 bleue, sinon icône usuelle. */
|
||||
function sharedFileIconSync(sharedPaths, itemName, vaultName, itemPath) {
|
||||
if (sharedPaths && sharedPaths.has(`${vaultName}::${itemPath}`)) return _sharedIcon();
|
||||
@@ -494,6 +657,12 @@ async function incrementalLoadDirectory(vaultName, dirPath, container) {
|
||||
}
|
||||
});
|
||||
|
||||
// Coffre Secrets (Phase 1c) : même nœud virtuel que dans loadDirectory.
|
||||
if (me && vaultName === `home-${me}` && !dirPath) {
|
||||
const pair = await _buildSecretsNode(vaultName);
|
||||
if (pair) fragment.insertBefore(pair, fragment.firstChild);
|
||||
}
|
||||
|
||||
// Replace container content in a single batch operation to avoid flash
|
||||
container.textContent = "";
|
||||
container.appendChild(fragment);
|
||||
@@ -838,6 +1007,14 @@ async function loadDirectory(vaultName, dirPath, container) {
|
||||
}
|
||||
});
|
||||
|
||||
// Coffre Secrets (Phase 1c) : nœud virtuel permanent en tête, à la racine
|
||||
// du Home de l'utilisateur courant seulement, jamais sous un autre Home
|
||||
// ni sous un vault de notes, et jamais comme un dossier physique.
|
||||
if (me && vaultName === `home-${me}` && !dirPath) {
|
||||
const pair = await _buildSecretsNode(vaultName);
|
||||
if (pair) fragment.insertBefore(pair, fragment.firstChild);
|
||||
}
|
||||
|
||||
container.appendChild(fragment);
|
||||
safeCreateIcons();
|
||||
}
|
||||
|
||||
@@ -2666,5 +2666,96 @@
|
||||
"guide105.dl_md_title": "Download this guide as Markdown",
|
||||
"guide105.dl_pdf_title": "Download this guide as PDF",
|
||||
"guide105.export_title": "ObsiGate User Guide",
|
||||
"guide105.export_footer": "Generated from ObsiGate {version} — {date}. This document mirrors the in-app guide; the latest version always lives in the application."
|
||||
"guide105.export_footer": "Generated from ObsiGate {version} — {date}. This document mirrors the in-app guide; the latest version always lives in the application.",
|
||||
"coffres.acces": "Access and consumers",
|
||||
"coffres.actions": "Actions",
|
||||
"coffres.active": "Active",
|
||||
"coffres.confirm_dek_rotate": "Rotate this vault's DEK? Secrets will be re-encrypted, with no change for consumers.",
|
||||
"coffres.confirm_delete": "Delete this secret? The following grants will break:",
|
||||
"coffres.confirm_revoke_consumer": "Revoke this consumer? All of its distributions will be denied.",
|
||||
"coffres.confirm_revoke_octroi": "Revoke this grant? Access will be denied on the next call.",
|
||||
"coffres.consumer_no_distribution": "No distribution in the recent log.",
|
||||
"coffres.consumers": "Consumers",
|
||||
"coffres.copied": "Copied.",
|
||||
"coffres.copy_failed": "Copy failed.",
|
||||
"coffres.copy_generated": "Copy the generated value",
|
||||
"coffres.copy_reference": "Copy",
|
||||
"coffres.create_coffre_title": "New vault",
|
||||
"coffres.create_octroi": "Create a grant",
|
||||
"coffres.create_secret_title": "New secret",
|
||||
"coffres.created": "Created",
|
||||
"coffres.dek_rotate": "Rotate DEK",
|
||||
"coffres.delete": "Delete",
|
||||
"coffres.deleted": "Secret deleted.",
|
||||
"coffres.description": "Description",
|
||||
"coffres.detail": "Details",
|
||||
"coffres.disabled": "Secret vaults are disabled on this instance (OBSIGATE_SECRETS_ENABLED).",
|
||||
"coffres.expiration": "Expiration",
|
||||
"coffres.expired": "Expired",
|
||||
"coffres.expires": "Expires",
|
||||
"coffres.expiring_soon": "Expiring soon",
|
||||
"coffres.generate": "Generate",
|
||||
"coffres.generate_unavailable": "Generation is unavailable in this browser.",
|
||||
"coffres.grant_all_coffre": "Whole vault",
|
||||
"coffres.journal": "Access log",
|
||||
"coffres.kek_absente": "Master key missing: metadata is visible, but no secret can be created or distributed.",
|
||||
"coffres.kek_generate": "Generate a KEK",
|
||||
"coffres.kek_generate_hint": "The generated KEK is returned only once and is never stored by ObsiGate: back it up outside the repository immediately.",
|
||||
"coffres.kek_new": "New KEK",
|
||||
"coffres.kek_ok": "Master key configured.",
|
||||
"coffres.kek_once_hint": "Copy it now and back it up outside the repository: it will never be shown again.",
|
||||
"coffres.kek_rotate": "Rotate KEK",
|
||||
"coffres.kek_rotated": "KEK rotated. Remember to replace the KEK in the environment or KEK file.",
|
||||
"coffres.kek_rotation_hint": "After rotation, replace the KEK in the environment or KEK file, otherwise distributions will fail.",
|
||||
"coffres.last_use": "Last use",
|
||||
"coffres.loading": "Loading vaults…",
|
||||
"coffres.manual_copy": "One-time manual copy",
|
||||
"coffres.manual_copy_admin_only": "Admin only (no session-based distribution endpoint exists yet on the backend).",
|
||||
"coffres.manual_copy_confirm": "A temporary consumer will be created, the value will be copied to the clipboard without being shown, then the consumer will be revoked immediately. The distribution will be audited.",
|
||||
"coffres.modified": "Modified",
|
||||
"coffres.name": "Name",
|
||||
"coffres.name_hint": "Lowercase, digits, - or _ ; 3 to 128 characters (e.g. gitea-token).",
|
||||
"coffres.name_required": "A name is required.",
|
||||
"coffres.never_in_note": "Never paste the value into the note.",
|
||||
"coffres.new_coffre": "New vault",
|
||||
"coffres.new_consumer": "New consumer",
|
||||
"coffres.new_secret": "New secret",
|
||||
"coffres.no_coffre": "No visible vault.",
|
||||
"coffres.no_secrets": "No secret in this vault.",
|
||||
"coffres.note": "Note",
|
||||
"coffres.octroi_created": "Grant created.",
|
||||
"coffres.octroi_default_90": "Empty = default expiration (90 days).",
|
||||
"coffres.octroi_no_expiration": "No expiration",
|
||||
"coffres.octroi_revoked": "Grant revoked.",
|
||||
"coffres.octrois": "Grants",
|
||||
"coffres.perimetre": "Scope",
|
||||
"coffres.propose_octroi": "Propose a grant (optional)",
|
||||
"coffres.reference": "Reference",
|
||||
"coffres.revoke": "Revoke",
|
||||
"coffres.rotated": "Secret rotated: the new version is active.",
|
||||
"coffres.rotation": "Rotate",
|
||||
"coffres.rotation_title": "Rotate secret",
|
||||
"coffres.save": "Save",
|
||||
"coffres.saved": "Secret saved. The value will no longer be shown.",
|
||||
"coffres.search_placeholder": "Filter secrets…",
|
||||
"coffres.secret": "Secret",
|
||||
"coffres.secret_saved_octroi_failed": "Secret saved, but the grant failed",
|
||||
"coffres.secrets": "Secrets",
|
||||
"coffres.select_secret": "Select a secret to see its metadata.",
|
||||
"coffres.service": "Service",
|
||||
"coffres.show_before_save": "Show before saving",
|
||||
"coffres.slug": "Slug",
|
||||
"coffres.slug_hint": "Lowercase, digits, - or _ ; 3 to 64 characters.",
|
||||
"coffres.status": "Status",
|
||||
"coffres.subtitle": "Secret vaults — metadata only, a value is never shown again after it is saved.",
|
||||
"coffres.target_id": "Target id",
|
||||
"coffres.target_id_required": "A target id is required.",
|
||||
"coffres.target_type": "Target type",
|
||||
"coffres.title": "Secrets",
|
||||
"coffres.token_once": "Token (shown only once)",
|
||||
"coffres.token_once_hint": "Copy it now: it will never be shown again.",
|
||||
"coffres.type": "Type",
|
||||
"coffres.value": "Value",
|
||||
"coffres.value_required": "A value is required.",
|
||||
"coffres.version": "Version"
|
||||
}
|
||||
|
||||
@@ -2666,5 +2666,96 @@
|
||||
"guide105.dl_md_title": "Télécharger ce guide en Markdown",
|
||||
"guide105.dl_pdf_title": "Télécharger ce guide en PDF",
|
||||
"guide105.export_title": "Guide d'utilisation ObsiGate",
|
||||
"guide105.export_footer": "Généré depuis ObsiGate {version} — {date}. Ce document est la copie du guide intégré ; la version la plus récente est toujours dans l'application."
|
||||
"guide105.export_footer": "Généré depuis ObsiGate {version} — {date}. Ce document est la copie du guide intégré ; la version la plus récente est toujours dans l'application.",
|
||||
"coffres.acces": "Accès et consommateurs",
|
||||
"coffres.actions": "Actions",
|
||||
"coffres.active": "Actif",
|
||||
"coffres.confirm_dek_rotate": "Pivoter la DEK de ce coffre ? Les secrets seront re-chiffrés, sans changement pour les consommateurs.",
|
||||
"coffres.confirm_delete": "Supprimer ce secret ? Les octrois suivants seront cassés :",
|
||||
"coffres.confirm_revoke_consumer": "Révoquer ce consommateur ? Toutes ses distributions seront refusées.",
|
||||
"coffres.confirm_revoke_octroi": "Révoquer cet octroi ? L'accès sera refusé dès le prochain appel.",
|
||||
"coffres.consumer_no_distribution": "Aucune distribution dans le journal récent.",
|
||||
"coffres.consumers": "Consommateurs",
|
||||
"coffres.copied": "Copié.",
|
||||
"coffres.copy_failed": "Copie impossible.",
|
||||
"coffres.copy_generated": "Copier la valeur générée",
|
||||
"coffres.copy_reference": "Copier",
|
||||
"coffres.create_coffre_title": "Nouveau coffre",
|
||||
"coffres.create_octroi": "Créer un octroi",
|
||||
"coffres.create_secret_title": "Nouveau secret",
|
||||
"coffres.created": "Créé",
|
||||
"coffres.dek_rotate": "Rotation DEK",
|
||||
"coffres.delete": "Supprimer",
|
||||
"coffres.deleted": "Secret supprimé.",
|
||||
"coffres.description": "Description",
|
||||
"coffres.detail": "Détail",
|
||||
"coffres.disabled": "Les coffres de secrets sont désactivés sur cette instance (OBSIGATE_SECRETS_ENABLED).",
|
||||
"coffres.expiration": "Expiration",
|
||||
"coffres.expired": "Expiré",
|
||||
"coffres.expires": "Expire",
|
||||
"coffres.expiring_soon": "Expire bientôt",
|
||||
"coffres.generate": "Générer",
|
||||
"coffres.generate_unavailable": "Génération indisponible dans ce navigateur.",
|
||||
"coffres.grant_all_coffre": "Tout le coffre",
|
||||
"coffres.journal": "Journal d'accès",
|
||||
"coffres.kek_absente": "Clé maître absente : les métadonnées sont visibles, mais aucun secret ne peut être créé ou distribué.",
|
||||
"coffres.kek_generate": "Générer une KEK",
|
||||
"coffres.kek_generate_hint": "La KEK générée n'est rendue qu'une fois et n'est jamais stockée par ObsiGate : sauvegardez-la hors dépôt immédiatement.",
|
||||
"coffres.kek_new": "Nouvelle KEK",
|
||||
"coffres.kek_ok": "Clé maître configurée.",
|
||||
"coffres.kek_once_hint": "Copiez-la maintenant et sauvegardez-la hors dépôt : elle ne sera plus jamais affichée.",
|
||||
"coffres.kek_rotate": "Rotation KEK",
|
||||
"coffres.kek_rotated": "KEK pivotée. Pensez à remplacer la KEK dans l'environnement ou le fichier KEK.",
|
||||
"coffres.kek_rotation_hint": "Après la rotation, remplacez la KEK dans l'environnement ou le fichier KEK, sinon les distributions échoueront.",
|
||||
"coffres.last_use": "Dernière utilisation",
|
||||
"coffres.loading": "Chargement des coffres…",
|
||||
"coffres.manual_copy": "Copie manuelle à usage unique",
|
||||
"coffres.manual_copy_admin_only": "Réservée aux administrateurs (aucun endpoint de distribution par session n'existe encore côté backend).",
|
||||
"coffres.manual_copy_confirm": "Un consommateur temporaire sera créé, la valeur sera copiée au presse-papiers sans être affichée, puis le consommateur sera révoqué immédiatement. La distribution sera auditée.",
|
||||
"coffres.modified": "Modifié",
|
||||
"coffres.name": "Nom",
|
||||
"coffres.name_hint": "Minuscules, chiffres, - ou _ ; 3 à 128 caractères (ex. gitea-token).",
|
||||
"coffres.name_required": "Un nom est requis.",
|
||||
"coffres.never_in_note": "Ne collez jamais la valeur dans la note.",
|
||||
"coffres.new_coffre": "Nouveau coffre",
|
||||
"coffres.new_consumer": "Nouveau consommateur",
|
||||
"coffres.new_secret": "Nouveau secret",
|
||||
"coffres.no_coffre": "Aucun coffre visible.",
|
||||
"coffres.no_secrets": "Aucun secret dans ce coffre.",
|
||||
"coffres.note": "Note",
|
||||
"coffres.octroi_created": "Octroi créé.",
|
||||
"coffres.octroi_default_90": "Vide = expiration par défaut (90 jours).",
|
||||
"coffres.octroi_no_expiration": "Sans expiration",
|
||||
"coffres.octroi_revoked": "Octroi révoqué.",
|
||||
"coffres.octrois": "Octrois",
|
||||
"coffres.perimetre": "Périmètre",
|
||||
"coffres.propose_octroi": "Proposer un octroi (optionnel)",
|
||||
"coffres.reference": "Référence",
|
||||
"coffres.revoke": "Révoquer",
|
||||
"coffres.rotated": "Secret pivoté : la nouvelle version est active.",
|
||||
"coffres.rotation": "Rotation",
|
||||
"coffres.rotation_title": "Rotation du secret",
|
||||
"coffres.save": "Enregistrer",
|
||||
"coffres.saved": "Secret enregistré. La valeur ne sera plus affichée.",
|
||||
"coffres.search_placeholder": "Filtrer les secrets…",
|
||||
"coffres.secret": "Secret",
|
||||
"coffres.secret_saved_octroi_failed": "Secret enregistré, mais l'octroi a échoué",
|
||||
"coffres.secrets": "Secrets",
|
||||
"coffres.select_secret": "Sélectionnez un secret pour voir ses métadonnées.",
|
||||
"coffres.service": "Service",
|
||||
"coffres.show_before_save": "Afficher avant l'enregistrement",
|
||||
"coffres.slug": "Identifiant (slug)",
|
||||
"coffres.slug_hint": "Minuscules, chiffres, - ou _ ; 3 à 64 caractères.",
|
||||
"coffres.status": "Statut",
|
||||
"coffres.subtitle": "Coffres de secrets — métadonnées seules, aucune valeur n'est jamais affichée après l'enregistrement.",
|
||||
"coffres.target_id": "Identifiant de la cible",
|
||||
"coffres.target_id_required": "L'identifiant de la cible est requis.",
|
||||
"coffres.target_type": "Type de cible",
|
||||
"coffres.title": "Secrets",
|
||||
"coffres.token_once": "Jeton (affiché une seule fois)",
|
||||
"coffres.token_once_hint": "Copiez-le maintenant : il ne sera plus jamais affiché.",
|
||||
"coffres.type": "Type",
|
||||
"coffres.value": "Valeur",
|
||||
"coffres.value_required": "Une valeur est requise.",
|
||||
"coffres.version": "Version"
|
||||
}
|
||||
|
||||
@@ -14331,3 +14331,35 @@ body.has-draco-bar .app-container { height: calc(100vh - 60px); }
|
||||
color: var(--text-primary); font: inherit; font-size: 13.5px;
|
||||
}
|
||||
.draco-liens-erreur { color: var(--err); font-size: 13px; margin: 8px 0 0; }
|
||||
|
||||
/* ---------------------------------------------------------------------------
|
||||
* Coffre Secrets (Phase 1c, issue #19) — vue Secrets + nœud sidebar.
|
||||
* Réutilise les variables de thème existantes ; aucune valeur de secret
|
||||
* n'est jamais affichée par ces styles (colonne valeur = •••••• en texte).
|
||||
* --------------------------------------------------------------------------- */
|
||||
.coffres-view { padding: 16px 20px; max-width: 1100px; }
|
||||
.coffres-subtitle { color: var(--text-muted); margin: 0 0 12px; }
|
||||
.coffres-banner { padding: 10px 12px; margin: 12px 0; background: var(--bg-secondary, #161b22); border: 1px solid var(--border, #21262d); border-radius: 8px; display: flex; gap: 8px; align-items: center; flex-wrap: wrap; }
|
||||
.coffres-banner.warn { border-color: var(--warning, #f59e0b); }
|
||||
.coffres-chips { display: flex; gap: 8px; flex-wrap: wrap; margin: 12px 0; }
|
||||
.coffres-chip { white-space: nowrap; }
|
||||
.coffres-actions { display: flex; gap: 8px; flex-wrap: wrap; align-items: center; margin: 12px 0; }
|
||||
.coffres-actions .modal-input { flex: 1 1 220px; }
|
||||
.coffres-table { width: 100%; border-collapse: collapse; margin: 12px 0; }
|
||||
.coffres-table th, .coffres-table td { text-align: left; padding: 8px 10px; border-bottom: 1px solid var(--border, #21262d); vertical-align: top; }
|
||||
.coffres-table tbody tr { cursor: pointer; }
|
||||
.coffres-table tbody tr:hover { background: var(--bg-hover, #1f2430); }
|
||||
.coffres-detail { margin-top: 16px; padding: 14px; background: var(--bg-secondary, #161b22); border: 1px solid var(--border, #21262d); border-radius: 8px; }
|
||||
.coffres-dl { display: grid; grid-template-columns: 160px 1fr; gap: 4px 12px; margin: 12px 0; }
|
||||
.coffres-dl dt { color: var(--text-muted); }
|
||||
.coffres-dl dd { margin: 0; overflow-wrap: anywhere; }
|
||||
.coffres-line { padding: 6px 0; border-bottom: 1px solid var(--border, #21262d); display: flex; gap: 8px; align-items: center; flex-wrap: wrap; justify-content: space-between; }
|
||||
.coffres-empty { color: var(--text-muted); }
|
||||
.coffres-alerts { margin: 12px 0; }
|
||||
.coffres-alert { color: var(--warning, #f59e0b); padding: 2px 0; }
|
||||
.coffres-form { display: grid; gap: 8px; max-width: 560px; margin: 8px 0; }
|
||||
.coffres-icon-lock { font-size: 16px; line-height: 1; }
|
||||
.coffres-alert-dot { color: var(--warning, #f59e0b); font-weight: 700; margin-left: 4px; }
|
||||
.coffres-context-menu { position: fixed; z-index: 1000; min-width: 220px; background: var(--bg-secondary, #161b22); border: 1px solid var(--border, #21262d); border-radius: 8px; padding: 4px; box-shadow: 0 8px 24px rgba(0, 0, 0, 0.4); }
|
||||
.coffres-context-item { display: block; width: 100%; text-align: left; padding: 8px 12px; background: transparent; border: none; color: var(--text-primary, #e6edf3); cursor: pointer; border-radius: 6px; }
|
||||
.coffres-context-item:hover { background: var(--bg-hover, #1f2430); }
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "obsigate",
|
||||
"version": "2.78.2",
|
||||
"version": "2.79.0",
|
||||
"description": "**Porte d'entrée web ultra-léger pour vos vaults Obsidian** — Accédez, naviguez et recherchez dans toutes vos notes Obsidian depuis n'importe quel appareil via une interface web moderne et responsive.",
|
||||
"main": "patch.js",
|
||||
"directories": {
|
||||
|
||||
@@ -0,0 +1,457 @@
|
||||
"""Tests du coffre Secrets ObsiGate (Phase 1a / 1b, issues #17 et #18).
|
||||
|
||||
Couverture exigée par les issues :
|
||||
- chiffrement/déchiffrement, tag invalide refusé, rotations (secret, DEK,
|
||||
KEK), mode dégradé sans KEK ;
|
||||
- matrice d'autorisation, y compris admin contre coffre personnel d'autrui
|
||||
et ACL « * » qui ne donne aucun droit sur un coffre ;
|
||||
- contexte manquant refusé, octroi révoqué refusé au prochain appel,
|
||||
expiration respectée, audit complet sans valeur ;
|
||||
- lectures = métadonnées seules, jamais de valeur ni de texte chiffré.
|
||||
|
||||
Toutes les valeurs de secret utilisées ici sont fictives et ne servent
|
||||
qu'aux tests. Le store est isolé par test via OBSIGATE_DATA_DIR et le
|
||||
journal d'audit via backend.audit.AUDIT_LOG_FILE.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from backend import audit as audit_module
|
||||
from backend.auth.middleware import require_admin, require_auth
|
||||
from backend.coffre_crypto import (
|
||||
KEKAbsenteError,
|
||||
KEKInvalideError,
|
||||
TagInvalideError,
|
||||
decrypt_secret_value,
|
||||
encrypt_secret_value,
|
||||
generate_dek,
|
||||
generate_kek_b64,
|
||||
load_kek,
|
||||
parse_kek,
|
||||
unwrap_dek,
|
||||
wrap_dek,
|
||||
)
|
||||
from backend.coffre_store import (
|
||||
CoffreExisteError,
|
||||
SecretExpireError,
|
||||
ValidationError,
|
||||
create_coffre,
|
||||
create_secret,
|
||||
distribuer_valeur,
|
||||
get_coffre_or_404,
|
||||
rotate_dek,
|
||||
rotate_kek,
|
||||
rotate_secret,
|
||||
)
|
||||
from backend.routers.coffres import router as coffres_router
|
||||
|
||||
VALEUR_TEST = "VALEUR-TEST-FICTIVE-7f3a9c-ne-pas-utiliser"
|
||||
VALEUR_TEST_2 = "VALEUR-TEST-FICTIVE-2-b4d8e1-ne-pas-utiliser"
|
||||
|
||||
# Utilisateur courant injecté dans les routes via dependency_overrides.
|
||||
_COURANT = {"user": None}
|
||||
|
||||
|
||||
def _fake_auth():
|
||||
if _COURANT["user"] is None:
|
||||
raise HTTPException(status_code=401, detail="Authentification requise")
|
||||
return _COURANT["user"]
|
||||
|
||||
|
||||
def _fake_admin():
|
||||
user = _fake_auth()
|
||||
if user.get("role") != "admin":
|
||||
raise HTTPException(status_code=403, detail="Accès admin requis")
|
||||
return user
|
||||
|
||||
|
||||
def _user(username: str, role: str = "user", vaults: list | None = None) -> dict:
|
||||
return {
|
||||
"username": username,
|
||||
"role": role,
|
||||
"vaults": vaults if vaults is not None else [f"home-{username}"],
|
||||
"active": True,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def env_coffre(tmp_path, monkeypatch):
|
||||
"""Environnement isolé : store, audit, KEK et feature flag."""
|
||||
monkeypatch.setenv("OBSIGATE_SECRETS_ENABLED", "1")
|
||||
monkeypatch.setenv("OBSIGATE_DATA_DIR", str(tmp_path))
|
||||
monkeypatch.setenv("OBSIGATE_KEK", generate_kek_b64())
|
||||
monkeypatch.delenv("OBSIGATE_KEK_FILE", raising=False)
|
||||
monkeypatch.setattr(audit_module, "AUDIT_LOG_FILE", tmp_path / "audit.log")
|
||||
_COURANT["user"] = _user("bruno", role="admin")
|
||||
yield tmp_path
|
||||
_COURANT["user"] = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(env_coffre):
|
||||
app = FastAPI()
|
||||
app.include_router(coffres_router)
|
||||
app.dependency_overrides[require_auth] = _fake_auth
|
||||
app.dependency_overrides[require_admin] = _fake_admin
|
||||
with TestClient(app, raise_server_exceptions=False) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
def _connecter(client, username: str, role: str = "user", vaults=None):
|
||||
_COURANT["user"] = _user(username, role=role, vaults=vaults)
|
||||
|
||||
|
||||
def _creer_coffre_secret(client, slug="personnel", nom="gitea-token", valeur=VALEUR_TEST):
|
||||
resp = client.post(
|
||||
"/api/coffres",
|
||||
json={"slug": slug, "nom": "Personnel", "perimetre": "personnel"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
resp = client.post(
|
||||
f"/api/coffres/{slug}/secrets",
|
||||
json={
|
||||
"nom": nom,
|
||||
"type": "token",
|
||||
"service": "Gitea",
|
||||
"valeur": valeur,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
return resp.json()
|
||||
|
||||
|
||||
def _creer_consommateur_octroi(client, slug="personnel", nom="gitea-token"):
|
||||
resp = client.post("/api/coffres/consommateurs", json={"nom": "spark"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
consommateur = resp.json()
|
||||
jeton = consommateur["jeton"]
|
||||
resp = client.post(
|
||||
f"/api/coffres/{slug}/octrois",
|
||||
json={
|
||||
"cible_type": "consommateur",
|
||||
"cible_id": consommateur["id"],
|
||||
"secret_nom": nom,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
return consommateur, jeton, resp.json()
|
||||
|
||||
|
||||
def _distribuer(client, jeton, reference="coffre:personnel/gitea-token",
|
||||
contexte_type="mission", contexte_id="mission-test-1"):
|
||||
return client.post(
|
||||
"/api/coffres/distribuer",
|
||||
json={
|
||||
"reference": reference,
|
||||
"contexte_type": contexte_type,
|
||||
"contexte_id": contexte_id,
|
||||
},
|
||||
headers={"Authorization": f"Bearer {jeton}"},
|
||||
)
|
||||
|
||||
|
||||
def _fichiers_sans_valeur(tmp_path, valeur=VALEUR_TEST):
|
||||
for path in tmp_path.iterdir():
|
||||
if path.is_file() and path.suffix in (".json", ".log"):
|
||||
assert valeur not in path.read_text(encoding="utf-8"), path.name
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Crypto (Phase 1a)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCrypto:
|
||||
def test_aller_retour_aes_gcm(self):
|
||||
kek = bytes(range(32))
|
||||
dek = generate_dek()
|
||||
wrapped = wrap_dek(dek, kek, "coffre-test")
|
||||
assert unwrap_dek(wrapped, kek, "coffre-test") == dek
|
||||
chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST)
|
||||
assert decrypt_secret_value(dek, "secret-1", 1, chiffre["nonce"], chiffre["ciphertext"]) == VALEUR_TEST
|
||||
|
||||
def test_tag_invalide_refuse(self):
|
||||
dek = generate_dek()
|
||||
chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST)
|
||||
alt = bytearray(base64.b64decode(chiffre["ciphertext"]))
|
||||
alt[-1] ^= 1
|
||||
chiffre_alt = {
|
||||
"nonce": chiffre["nonce"],
|
||||
"ciphertext": base64.b64encode(bytes(alt)).decode("ascii"),
|
||||
}
|
||||
with pytest.raises(TagInvalideError):
|
||||
decrypt_secret_value(dek, "secret-1", 1, chiffre_alt["nonce"], chiffre_alt["ciphertext"])
|
||||
|
||||
def test_aad_incorrect_refuse(self):
|
||||
dek = generate_dek()
|
||||
chiffre = encrypt_secret_value(dek, "secret-1", 1, VALEUR_TEST)
|
||||
with pytest.raises(TagInvalideError):
|
||||
decrypt_secret_value(dek, "secret-2", 1, chiffre["nonce"], chiffre["ciphertext"])
|
||||
|
||||
def test_parse_kek_invalide(self, monkeypatch):
|
||||
with pytest.raises(KEKInvalideError):
|
||||
parse_kek("trop-court")
|
||||
assert len(parse_kek(generate_kek_b64())) == 32
|
||||
|
||||
def test_kek_absente(self, tmp_path, monkeypatch):
|
||||
monkeypatch.delenv("OBSIGATE_KEK", raising=False)
|
||||
monkeypatch.delenv("OBSIGATE_KEK_FILE", raising=False)
|
||||
with pytest.raises(KEKAbsenteError):
|
||||
load_kek()
|
||||
|
||||
def test_kek_fichier_0600_ok_et_trop_permissif_refuse(self, tmp_path, monkeypatch):
|
||||
kek_file = tmp_path / "kek.txt"
|
||||
kek_file.write_text(generate_kek_b64(), encoding="utf-8")
|
||||
os.chmod(kek_file, 0o600)
|
||||
monkeypatch.delenv("OBSIGATE_KEK", raising=False)
|
||||
monkeypatch.setenv("OBSIGATE_KEK_FILE", str(kek_file))
|
||||
assert len(load_kek()) == 32
|
||||
os.chmod(kek_file, 0o644)
|
||||
with pytest.raises(KEKInvalideError):
|
||||
load_kek()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Store (Phase 1a)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestStore:
|
||||
def test_roundtrip_et_aucun_clair_au_repos(self, env_coffre):
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST, service="Gitea")
|
||||
assert distribuer_valeur(coffre, secret) == VALEUR_TEST
|
||||
_fichiers_sans_valeur(env_coffre)
|
||||
|
||||
def test_rotation_secret(self, env_coffre):
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST)
|
||||
secret = rotate_secret(coffre, "gitea-token", VALEUR_TEST_2)
|
||||
assert secret["version_courante"] == 2
|
||||
assert distribuer_valeur(coffre, secret) == VALEUR_TEST_2
|
||||
|
||||
def test_rotation_dek_puis_kek(self, env_coffre, monkeypatch):
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
secret = create_secret(coffre, "gitea-token", "token", VALEUR_TEST)
|
||||
rotate_dek("personnel")
|
||||
assert distribuer_valeur(get_coffre_or_404("personnel"), secret) == VALEUR_TEST
|
||||
nouvelle_kek = generate_kek_b64()
|
||||
assert rotate_kek(nouvelle_kek) == 1
|
||||
monkeypatch.setenv("OBSIGATE_KEK", nouvelle_kek)
|
||||
assert distribuer_valeur(get_coffre_or_404("personnel"), secret) == VALEUR_TEST
|
||||
|
||||
def test_mode_degrade_sans_kek(self, env_coffre, monkeypatch):
|
||||
monkeypatch.delenv("OBSIGATE_KEK", raising=False)
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
assert coffre["dek_chiffree"] is None
|
||||
with pytest.raises(KEKAbsenteError):
|
||||
create_secret(coffre, "gitea-token", "token", VALEUR_TEST)
|
||||
|
||||
def test_validation_et_doublons(self, env_coffre):
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
with pytest.raises(ValidationError):
|
||||
create_coffre("MAUVAIS SLUG !!", "X", "bruno")
|
||||
with pytest.raises(CoffreExisteError):
|
||||
create_coffre("personnel", "Autre", "bruno")
|
||||
create_secret(coffre, "gitea-token", "token", VALEUR_TEST)
|
||||
with pytest.raises(ValidationError):
|
||||
create_secret(
|
||||
coffre, "autre-secret", "token", VALEUR_TEST,
|
||||
note="api-key: ceci-ressemble-a-un-secret",
|
||||
)
|
||||
|
||||
def test_secret_expire_refuse_a_la_distribution(self, env_coffre):
|
||||
coffre = create_coffre("personnel", "Personnel", "bruno")
|
||||
hier = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
secret = create_secret(
|
||||
coffre, "vieux-jeton", "token", VALEUR_TEST, expiration=hier
|
||||
)
|
||||
with pytest.raises(SecretExpireError):
|
||||
distribuer_valeur(coffre, secret)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# API et autorisations (Phase 1b)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestApiAutorisations:
|
||||
def test_flag_desactive_repond_503(self, client, monkeypatch):
|
||||
monkeypatch.setenv("OBSIGATE_SECRETS_ENABLED", "0")
|
||||
assert client.get("/api/coffres").status_code == 503
|
||||
|
||||
def test_lectures_metadonnees_seules(self, client, env_coffre):
|
||||
_creer_coffre_secret(client)
|
||||
resp = client.get("/api/coffres/personnel/secrets")
|
||||
assert resp.status_code == 200
|
||||
corps = resp.text
|
||||
assert VALEUR_TEST not in corps
|
||||
assert "ciphertext" not in corps
|
||||
assert "nonce" not in corps
|
||||
item = resp.json()[0]
|
||||
assert item["has_value"] is True
|
||||
assert item["reference"] == "coffre:personnel/gitea-token"
|
||||
assert "valeur" not in item
|
||||
|
||||
def test_admin_ne_voit_pas_le_coffre_personnel_d_autrui(self, client):
|
||||
_creer_coffre_secret(client) # propriétaire : bruno (admin de test)
|
||||
_connecter(client, "alice", role="user")
|
||||
assert client.get("/api/coffres/personnel").status_code == 403
|
||||
assert client.get("/api/coffres/personnel/secrets").status_code == 403
|
||||
# Même un admin avec vaults=["*"] n'a aucun droit sur un coffre
|
||||
# personnel dont il n'est pas propriétaire.
|
||||
_connecter(client, "admin2", role="admin", vaults=["*"])
|
||||
assert client.get("/api/coffres/personnel").status_code == 403
|
||||
|
||||
def test_acl_etoile_ne_donne_aucun_droit_coffre(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_connecter(client, "mallory", role="user", vaults=["*"])
|
||||
assert client.get("/api/coffres/personnel/secrets").status_code == 403
|
||||
|
||||
def test_coffre_instance_reserve_admin(self, client):
|
||||
_connecter(client, "bruno", role="user")
|
||||
resp = client.post(
|
||||
"/api/coffres",
|
||||
json={"slug": "instance", "nom": "Instance", "perimetre": "instance"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
_connecter(client, "bruno", role="admin")
|
||||
resp = client.post(
|
||||
"/api/coffres",
|
||||
json={"slug": "instance", "nom": "Instance", "perimetre": "instance"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
_connecter(client, "alice", role="user")
|
||||
assert client.get("/api/coffres/instance").status_code == 403
|
||||
|
||||
def test_creation_secret_sans_kek_repond_503(self, client, monkeypatch):
|
||||
monkeypatch.delenv("OBSIGATE_KEK", raising=False)
|
||||
resp = client.post("/api/coffres", json={"slug": "personnel", "nom": "Personnel"})
|
||||
assert resp.status_code == 200 # métadonnées créées en mode dégradé
|
||||
resp = client.post(
|
||||
"/api/coffres/personnel/secrets",
|
||||
json={"nom": "gitea-token", "type": "token", "valeur": VALEUR_TEST},
|
||||
)
|
||||
assert resp.status_code == 503
|
||||
assert resp.json()["detail"] == "kek_absente"
|
||||
|
||||
|
||||
class TestDistribution:
|
||||
def test_distribution_ok_puis_revoquee_puis_consommateur_revoque(self, client, env_coffre):
|
||||
_creer_coffre_secret(client)
|
||||
consommateur, jeton, octroi = _creer_consommateur_octroi(client)
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["valeur"] == VALEUR_TEST
|
||||
assert resp.json()["version"] == 1
|
||||
# Le jeton complet n'est jamais relisté, ni son hash (seul le
|
||||
# préfixe d'affichage, non exploitable, apparaît).
|
||||
liste = client.get("/api/coffres/consommateurs").json()
|
||||
assert jeton not in json.dumps(liste)
|
||||
assert "jeton_hash" not in json.dumps(liste)
|
||||
assert all(set(item.keys()) == {"id", "nom", "jeton_prefixe", "statut", "cree_par", "cree_le"} for item in liste)
|
||||
# Révocation de l'octroi : refusé dès le prochain appel.
|
||||
resp = client.delete(f"/api/coffres/octrois/{octroi['id']}")
|
||||
assert resp.status_code == 200
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["detail"] == "octroi_revoque"
|
||||
# Révocation du consommateur : tous ses octrois tombent.
|
||||
resp = client.post(
|
||||
"/api/coffres/personnel/octrois",
|
||||
json={"cible_type": "consommateur", "cible_id": consommateur["id"], "secret_nom": "gitea-token"},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
resp = client.delete(f"/api/coffres/consommateurs/{consommateur['id']}")
|
||||
assert resp.status_code == 200
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["detail"] == "consommateur_revoque"
|
||||
|
||||
def test_contexte_manquant_refuse(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_, jeton, _ = _creer_consommateur_octroi(client)
|
||||
resp = client.post(
|
||||
"/api/coffres/distribuer",
|
||||
json={"reference": "coffre:personnel/gitea-token"},
|
||||
headers={"Authorization": f"Bearer {jeton}"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["detail"] == "contexte_manquant"
|
||||
|
||||
def test_session_utilisateur_ne_distribue_pas(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_creer_consommateur_octroi(client)
|
||||
resp = _distribuer(client, "jeton-session-utilisateur-fictif")
|
||||
assert resp.status_code == 401
|
||||
assert resp.json()["detail"] == "consommateur_invalide"
|
||||
|
||||
def test_octroi_absent_et_secret_expire(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
resp = client.post("/api/coffres/consommateurs", json={"nom": "sans-octroi"})
|
||||
jeton = resp.json()["jeton"]
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["detail"] == "octroi_absent"
|
||||
# Secret expiré : 410, même avec un octroi actif.
|
||||
_connecter(client, "bruno", role="admin")
|
||||
hier = (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()
|
||||
resp = client.post(
|
||||
"/api/coffres/personnel/secrets",
|
||||
json={"nom": "vieux-jeton", "type": "token", "valeur": VALEUR_TEST, "expiration": hier},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
_, jeton2, _ = _creer_consommateur_octroi(client, nom="vieux-jeton")
|
||||
resp = _distribuer(client, jeton2, reference="coffre:personnel/vieux-jeton")
|
||||
assert resp.status_code == 410
|
||||
assert resp.json()["detail"] == "secret_expire"
|
||||
|
||||
def test_rotation_change_la_valeur_distribuee(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_, jeton, _ = _creer_consommateur_octroi(client)
|
||||
resp = client.post(
|
||||
"/api/coffres/personnel/secrets/gitea-token/rotation",
|
||||
json={"valeur": VALEUR_TEST_2},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["version"] == 2
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.json()["valeur"] == VALEUR_TEST_2
|
||||
|
||||
def test_suppression_casse_les_octrois(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_, jeton, _ = _creer_consommateur_octroi(client)
|
||||
resp = client.delete("/api/coffres/personnel/secrets/gitea-token")
|
||||
assert resp.status_code == 200
|
||||
resp = _distribuer(client, jeton)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["detail"] == "octroi_absent"
|
||||
|
||||
def test_octroi_par_defaut_90_jours(self, client):
|
||||
_creer_coffre_secret(client)
|
||||
_, _, octroi = _creer_consommateur_octroi(client)
|
||||
expiration = datetime.fromisoformat(octroi["expiration"])
|
||||
delta = expiration - datetime.now(timezone.utc)
|
||||
assert 89 <= delta.days <= 91
|
||||
|
||||
def test_audit_sans_valeur_et_journal(self, client, env_coffre):
|
||||
_creer_coffre_secret(client)
|
||||
_, jeton, _ = _creer_consommateur_octroi(client)
|
||||
assert _distribuer(client, jeton).status_code == 200
|
||||
_fichiers_sans_valeur(env_coffre)
|
||||
audit = (env_coffre / "audit.log").read_text(encoding="utf-8")
|
||||
assert "distribution_ok" in audit
|
||||
assert VALEUR_TEST not in audit
|
||||
resp = client.get("/api/coffres/personnel/journal")
|
||||
assert resp.status_code == 200
|
||||
actions = [e["action"] for e in resp.json()]
|
||||
assert "secret_cree" in actions
|
||||
assert "distribution_ok" in actions
|
||||
assert "valeur" not in json.dumps(resp.json())
|
||||
@@ -129,3 +129,20 @@ def test_rail_mobile_vers_le_vrai_hamburger():
|
||||
assert 'matchMedia("(max-width: 768px)")' in html
|
||||
assert '"hamburger-btn"' in html
|
||||
assert '"sidebar-toggle-btn"' in html
|
||||
|
||||
|
||||
def test_panneau_version_service_affichee_225():
|
||||
# #225 : le pied du panneau Spark (mode sidebar) affiche la
|
||||
# version du service d'agents, comme la console pleine page.
|
||||
# La version est lue en direct dans la santé du service
|
||||
# (GET /healthz de l'origine injectée par l'hôte), à l'ouverture
|
||||
# du panneau ; jamais un littéral de version dans l'asset, et
|
||||
# source injoignable = ligne laissée masquée (rien d'affiché).
|
||||
panneau = (FRONTEND / "draco-panneau.js").read_text(encoding="utf-8")
|
||||
assert "draco-panneau-version" in panneau
|
||||
assert "chargerVersion" in panneau
|
||||
assert 'API + "/healthz"' in panneau
|
||||
assert '"Service d\'agents v" + versionConnue' in panneau
|
||||
# La ligne naît masquée et n'est démasquée qu'avec une version lue.
|
||||
assert 'class="draco-panneau-version"' in panneau
|
||||
assert "versionEl.hidden = false" in panneau
|
||||
|
||||
Reference in New Issue
Block a user