diff --git a/backend/coffre_migration.py b/backend/coffre_migration.py new file mode 100644 index 0000000..edab6fa --- /dev/null +++ b/backend/coffre_migration.py @@ -0,0 +1,306 @@ +"""Migration Phase 2 du coffre Secrets ObsiGate (issue #20). + +Bascule des clés aujourd'hui en clair dans ``data/api_keys.json`` vers le +coffre d'instance « ObsiGate interne » (slug ``obsigate-interne``). + +Format source (constaté dans ``backend/ai.py`` et +``backend/tools/secrets.py``) : un dictionnaire plat ``{NOM: valeur}``. +Seules les entrées dont le nom ressemble à un secret sont migrables : +clés IA et clés d'outils. Les valeurs non secrètes (par exemple +``OBSIGATE_GITEA_URL``) ne sont JAMAIS migrées et ne sont jamais +touchées par la purge. + +Règles de sécurité, non négociables : +- Aucune fonction de ce module ne renvoie une valeur, ne la journalise, + ni ne l'écrit ailleurs que dans le coffre (chiffrée, via le store). + Les aperçus ne contiennent que des noms, des types et ``has_value``. +- L'import ne supprime rien : il crée des secrets dans le coffre et + laisse ``api_keys.json`` intact. La purge est une étape séparée, qui + exige ``confirmer=True`` et ne supprime que les noms listés. +- Un secret déjà présent dans le coffre n'est JAMAIS écrasé : il est + signalé dans ``deja_existants`` et la valeur du coffre est conservée. + +Limite connue, documentée dans ``docs/COFFRE-sauvegarde-restauration.md`` : +les lecteurs actuels (``backend/ai.py``, ``backend/tools/secrets.py``) +lisent encore ``api_keys.json`` (chemin relatif ``data/``, sans passer +par ``OBSIGATE_DATA_DIR``). Purger le clair sans avoir branché le +lecteur concerné sur le coffre rend donc la clé inaccessible à ce +lecteur : la purge reste un choix explicite de l'opérateur, confirmé +deux fois côté interface (import, puis purge). +""" + +from __future__ import annotations + +import json +import logging +import os +import threading +from pathlib import Path +from typing import Any + +from backend.coffre_store import ( + ValidationError, + create_coffre, + create_secret, + get_coffre, + get_secret, + resolve_data_dir, +) + +logger = logging.getLogger("obsigate.coffre_migration") + +_lock = threading.RLock() + +COFFRE_INTERNE_SLUG = "obsigate-interne" +COFFRE_INTERNE_NOM = "ObsiGate interne" + +# (nom dans api_keys.json, type de secret, service d'affichage). +# Miroir des clés lues par backend/ai.py (clés IA) et par +# backend/tools/secrets.py TOOL_KEY_NAMES (clés d'outils). L'URL Gitea +# est volontairement absente : ce n'est pas un secret. +CLES_MIGRABLES: tuple[tuple[str, str, str], ...] = ( + ("DEEPSEEK_API_KEY", "api_key", "DeepSeek"), + ("OPENROUTER_API_KEY", "api_key", "OpenRouter"), + ("GEMINI_API_KEY", "api_key", "Gemini"), + ("NVIDIA_API_KEY", "api_key", "NVIDIA"), + ("QWENCLOUD_API_KEY", "api_key", "QwenCloud"), + ("XIAOMI_API_KEY", "api_key", "Xiaomi"), + ("MISTRAL_API_KEY", "api_key", "Mistral"), + ("OBSIGATE_TAVILY_API_KEY", "api_key", "Tavily"), + ("OBSIGATE_BRAVE_API_KEY", "api_key", "Brave"), + ("OBSIGATE_SERPAPI_API_KEY", "api_key", "SerpAPI"), + ("OBSIGATE_EXA_API_KEY", "api_key", "Exa"), + ("OBSIGATE_GITEA_TOKEN", "token", "Gitea"), + ("OBSIGATE_GITHUB_TOKEN", "token", "GitHub"), +) + +_NOMS_MIGRABLES = {nom for nom, _, _ in CLES_MIGRABLES} + + +def _nom_secret(nom_source: str) -> str: + """Nom de secret dérivé du nom source (minuscules, valide NOM_SECRET_RE).""" + return nom_source.lower() + + +def _api_keys_path(base: Path) -> Path: + return base / "api_keys.json" + + +def _read_api_keys(base: Path) -> dict: + path = _api_keys_path(base) + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + logger.warning("api_keys.json illisible (%s) : %s", path, exc) + return {} + return data if isinstance(data, dict) else {} + + +def _write_api_keys(base: Path, data: dict) -> None: + """Écriture atomique tmp+rename, fichier final en 0600 (best-effort).""" + path = _api_keys_path(base) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_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 lister_candidats(data_dir: str | Path | None = None) -> list: + """Aperçu de migration : métadonnées seules, jamais de valeur. + + Chaque candidat porte ``nom`` (nom dans le coffre), ``source`` (nom + dans api_keys.json), ``type``, ``service``, ``has_value`` (une valeur + non vide est présente dans le fichier) et ``deja_existant`` (un + secret de ce nom existe déjà dans le coffre d'instance). + """ + base = resolve_data_dir(data_dir) + with _lock: + cles = _read_api_keys(base) + coffre = get_coffre(COFFRE_INTERNE_SLUG, base) + candidats = [] + for source, type_secret, service in CLES_MIGRABLES: + nom = _nom_secret(source) + valeur = cles.get(source) + candidats.append( + { + "nom": nom, + "source": source, + "type": type_secret, + "service": service, + "has_value": isinstance(valeur, str) and bool(valeur.strip()), + "deja_existant": ( + get_secret(coffre["id"], nom, base) is not None + if coffre is not None + else False + ), + } + ) + return candidats + + +def importer_api_keys( + data_dir: str | Path | None = None, cree_par: str = "" +) -> dict: + """Importe les entrées avec valeur vers le coffre d'instance. + + Ne supprime rien dans ``api_keys.json``. Retourne uniquement des + noms et des comptes : ``migres``, ``deja_existants``, ``sans_valeur``. + Lève ``KEKAbsenteError`` / ``CryptoIndisponibleError`` si le coffre + ne peut pas chiffrer (le routeur les traduit en 503). + """ + base = resolve_data_dir(data_dir) + with _lock: + cles = _read_api_keys(base) + coffre = get_coffre(COFFRE_INTERNE_SLUG, base) + if coffre is None: + coffre = create_coffre( + slug=COFFRE_INTERNE_SLUG, + nom=COFFRE_INTERNE_NOM, + proprietaire=(cree_par or "admin").strip(), + perimetre="instance", + description="Coffre d'instance : clés IA et clés d'outils migrées depuis api_keys.json (Phase 2).", + data_dir=base, + ) + migres: list[str] = [] + deja_existants: list[str] = [] + sans_valeur: list[str] = [] + for source, type_secret, service in CLES_MIGRABLES: + nom = _nom_secret(source) + valeur = cles.get(source) + if not (isinstance(valeur, str) and valeur.strip()): + sans_valeur.append(nom) + continue + if get_secret(coffre["id"], nom, base) is not None: + deja_existants.append(nom) + continue + create_secret( + coffre, + nom, + type_secret, + valeur, + service=service, + note="Migré depuis api_keys.json (Phase 2).", + data_dir=base, + ) + migres.append(nom) + return { + "coffre": COFFRE_INTERNE_SLUG, + "migres": migres, + "deja_existants": deja_existants, + "sans_valeur": sans_valeur, + } + + +def purger_api_keys( + noms: list[str] | None, + data_dir: str | Path | None = None, + confirmer: bool = False, +) -> dict: + """Supprime du clair les noms listés, après confirmation explicite. + + ``confirmer`` doit valoir True, sinon refus (jamais automatique). + Seuls les noms migrables sont acceptés ; un nom inconnu refuse + l'opération entière, pour ne jamais supprimer une clé par erreur. + Retourne ``supprimes`` et ``absents`` (noms, jamais de valeurs). + """ + if not confirmer: + raise ValidationError( + "Suppression du clair refusée : confirmation explicite requise" + ) + demandes = [(n or "").strip().lower() for n in (noms or [])] + # Accepte indifféremment le nom source (DEEPSEEK_API_KEY) ou le nom + # de secret (deepseek_api_key) ; tout le reste refuse l'opération. + noms_secrets = {_nom_secret(source) for source in _NOMS_MIGRABLES} + inconnus = [ + n + for n in demandes + if n.upper() not in _NOMS_MIGRABLES and n not in noms_secrets + ] + if inconnus: + raise ValidationError(f"Noms inconnus, purge annulée : {', '.join(inconnus)}") + sources_a_supprimer = set() + for demande in demandes: + for source, _, _ in CLES_MIGRABLES: + if demande in (source.lower(), _nom_secret(source)): + sources_a_supprimer.add(source) + base = resolve_data_dir(data_dir) + with _lock: + cles = _read_api_keys(base) + supprimes: list[str] = [] + absents: list[str] = [] + for source in sorted(sources_a_supprimer): + if source in cles: + del cles[source] + supprimes.append(_nom_secret(source)) + else: + absents.append(_nom_secret(source)) + _write_api_keys(base, cles) + return {"supprimes": supprimes, "absents": absents} + + +def creer_consommateur_avec_octrois( + nom: str, + octrois_demandes: list[dict[str, Any]] | None, + cree_par: str, + data_dir: str | Path | None = None, +) -> tuple[dict, str, list, list]: + """Enregistre un consommateur et ses octrois en une seule opération. + + Retourne (consommateur public, jeton à usage unique, octrois créés, + erreurs). Une erreur sur un octroi (coffre ou secret introuvable) + n'annule pas le consommateur ni les autres octrois : elle est + signalée dans ``erreurs`` (noms seuls, jamais de valeurs). Le coffre + lui-même n'est jamais modifié par cette opération. + """ + from backend.coffre_store import ( # import local : évite un cycle + IntrouvableError, + create_consommateur, + create_octroi, + ) + + base = resolve_data_dir(data_dir) + consommateur, jeton = create_consommateur(nom, cree_par, base) + crees: list[dict] = [] + erreurs: list[dict] = [] + for demande in octrois_demandes or []: + slug = str(demande.get("coffre_slug", "") or "") + secret_nom = str(demande.get("secret_nom", "") or "") or None + try: + coffre = get_coffre(slug, base) + if coffre is None: + raise IntrouvableError(f"Coffre introuvable : {slug}") + secret_id = None + if secret_nom is not None: + secret = get_secret(coffre["id"], secret_nom, base) + if secret is None: + raise IntrouvableError(f"Secret introuvable : {secret_nom}") + secret_id = secret["id"] + octroi = create_octroi( + coffre=coffre, + cible_type="consommateur", + cible_id=consommateur["id"], + secret_id=secret_id, + cree_par=cree_par, + expiration=demande.get("expiration"), + data_dir=base, + ) + crees.append(octroi) + except (IntrouvableError, ValidationError) as exc: + erreurs.append( + { + "coffre_slug": slug, + "secret_nom": secret_nom, + "erreur": str(exc), + } + ) + return consommateur, jeton, crees, erreurs