Set-OPS-Public/scripts/proxmox_api.py

#!/usr/bin/env python3
"""Acces au cluster Proxmox de l'hebergeur — partage par les applicateurs.

Extrait pour n'avoir QU'UN endroit ou vit la recomposition du jeton : la voute ne
porte que le NOM du jeton, l'utilisateur vit dans `proxmox-hebergeur.yml`. Les
recomposer a deux endroits, c'est se preparer un 401 muet le jour ou l'un des deux
change.
"""
from __future__ import annotations

import json
import os
import ssl
import subprocess
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

import yaml

RACINE = Path(__file__).resolve().parent.parent


def hebergeur() -> Path:
    """Depot de l'hebergeur, designe par le symlink `underlay.yml` (D-17)."""
    lien = Path(os.environ.get("SETOPS_UNDERLAY") or (RACINE / "underlay.yml"))
    if not lien.exists():
        raise SystemExit("Aucun underlay ne designe d'hebergeur : pas de cluster a piloter.")
    return lien.resolve().parent


def _lire_voute(p: Path) -> dict:
    r = subprocess.run(["ansible-vault", "view", str(p)], capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit("Voute illisible : renseigner ANSIBLE_VAULT_PASSWORD_FILE.\n"
                         + r.stderr.strip()[:300])
    return yaml.safe_load(r.stdout) or {}


def voute_underlay(base: Path) -> dict:
    """Les secrets du MONDE PHYSIQUE : `underlay.vault.yml`, chez l'hebergeur.

    LA VOUTE DE L'UNDERLAY EST DISTINCTE DE CELLE DES TENANTS (2026-08-22). Le jeton d'API
    du cluster et la cle de la frontiere appartiennent a l'hebergeur, pas a l'organisation
    hebergee. Recopies dans la voute de chaque tenant, ils ne pouvaient plus etre revoques
    isolement : retirer l'acces d'un locataire revenait a le retirer a tous.

    C'est la faute des neuf copies de la resolution d'instance, appliquee aux secrets. On
    ferme la meme porte, du meme geste : une source, pas N copies.

    Rend {} si le fichier n'existe pas — l'appelant retombe alors sur la voute du tenant,
    le temps que les sites migrent.
    """
    p = base / "underlay.vault.yml"
    return _lire_voute(p) if p.is_file() else {}


def voute(base: Path) -> dict:
    """Secrets, lus par ansible-vault. Rien n'est journalise ni ecrit sur disque.

    L'ordre compte : la voute de l'UNDERLAY d'abord (monde physique), celle du tenant
    ensuite. Un site migre est donc lu correctement, un site qui ne l'est pas encore
    continue de fonctionner — et le jour ou les deux portent la cle, c'est celle de
    l'hebergeur qui fait foi, ce qui est le sens de la frontiere.
    """
    fusion: dict = {}
    for nom in ("principal", "production", "lab"):
        p = base / "inventories" / nom / "group_vars" / "all" / "vault.yml"
        if p.is_file():
            fusion = _lire_voute(p)
            break
    physique = voute_underlay(base)
    if not fusion and not physique:
        raise SystemExit(f"Aucune voute sous {base} (ni underlay.vault.yml, "
                         f"ni inventories/*/group_vars/all/vault.yml).")
    fusion.update(physique)
    return fusion


class Cluster:
    """Le minimum d'API Proxmox. Retourne `{'_erreur': ...}` plutot que de lever."""

    def __init__(self, hote: str, port, user: str, tid: str, secret: str):
        self.base = f"https://{hote}:{port}/api2/json"
        self.hote = hote
        self.tok = f"PVEAPIToken={user}!{str(tid).split('!')[-1]}={secret}"
        self.ctx = ssl._create_unverified_context()

    @classmethod
    def depuis_hebergeur(cls) -> tuple["Cluster", Path]:
        base = hebergeur()
        heb = yaml.safe_load((base / "proxmox-hebergeur.yml").read_text(encoding="utf-8")) or {}
        v = voute(base)
        return cls(heb.get("proxmox_api_host"), heb.get("proxmox_api_port", 8006),
                   heb.get("proxmox_api_user"), v["proxmox_api_token_id"],
                   v["proxmox_api_token_secret"]), base

    def __call__(self, chemin: str, methode: str = "GET", corps: dict | None = None):
        data = urllib.parse.urlencode(corps, doseq=True).encode() if corps else None
        req = urllib.request.Request(self.base + chemin, data=data, method=methode,
                                     headers={"Authorization": self.tok})
        try:
            with urllib.request.urlopen(req, context=self.ctx, timeout=45) as rep:
                return json.loads(rep.read()).get("data")
        except urllib.error.HTTPError as e:
            return {"_erreur": f"{e.code} {e.read().decode()[:200]}"}
        except OSError as e:
            return {"_erreur": str(e)[:200]}

    def rate(self, rep) -> str | None:
        return rep["_erreur"] if isinstance(rep, dict) and rep.get("_erreur") else None


def ssh(hote: str, commande: str, entree: str | None = None) -> tuple[int, str]:
    """SSH vers un noeud. La cle vient de l'agent, ou de SETOPS_SSH_KEY."""
    cmd = ["ssh", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=no",
           "-o", "ConnectTimeout=10"]
    cle = os.environ.get("SETOPS_SSH_KEY")
    if cle:
        cmd += ["-o", "IdentitiesOnly=yes", "-i", os.path.expanduser(cle)]
    cmd += [f"{os.environ.get('SETOPS_SSH_USER', 'ansible')}@{hote}", commande]
    r = subprocess.run(cmd, input=entree, capture_output=True, text=True, timeout=90)
    return r.returncode, (r.stdout or "") + (r.stderr or "")