ObsiGate/backend/coffre_migration.py

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