Set-OPS/scripts/config_proxmox.py

#!/usr/bin/env python3
"""Assistant de configuration Proxmox pour Set-OPS."""

from __future__ import annotations

from getpass import getpass
from pathlib import Path
import os
import subprocess
import sys

import yaml


RACINE = Path(__file__).resolve().parents[1]
INSTANCE = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance"))
FICHIER_PROXMOX = INSTANCE / "inventories/lab/group_vars/proxmox.yml"
FICHIER_VAULT = INSTANCE / "inventories/lab/group_vars/proxmox.vault.yml"


VALEURS_DEFAUT = {
    "proxmox_api_host": "",
    "proxmox_api_user": "",
    "proxmox_api_port": "",
    "proxmox_validate_certs": False,
    "proxmox_clone_noeud": "",
    "proxmox_clone_vmid_modele": 9000,
    "proxmox_clone_source_nom": "modele-debian13",
    "proxmox_clone_stockage": "",
    "proxmox_clone_format": "",
    "proxmox_clone_complet": True,
    "proxmox_clone_timeout": 600,
    "proxmox_clone_disque": "scsi0",
    "proxmox_clone_interface": "net0",
    "proxmox_clone_pont": "vmbr0",
    "proxmox_clone_parefeu_interface": False,
    "proxmox_clone_demarrer": True,
}


LIBELLES = {
    "proxmox_api_host": "Hote API Proxmox",
    "proxmox_api_user": "Utilisateur API Proxmox",
    "proxmox_api_port": "Port API Proxmox",
    "proxmox_validate_certs": "Valider les certificats TLS",
    "proxmox_clone_noeud": "Noeud Proxmox par defaut",
    "proxmox_clone_vmid_modele": "VMID du modele Debian 13",
    "proxmox_clone_source_nom": "Nom logique du modele",
    "proxmox_clone_stockage": "Stockage Proxmox par defaut",
    "proxmox_clone_format": "Format disque par defaut",
    "proxmox_clone_complet": "Clone complet",
    "proxmox_clone_timeout": "Timeout operations Proxmox",
    "proxmox_clone_disque": "Disque principal",
    "proxmox_clone_interface": "Interface reseau",
    "proxmox_clone_pont": "Pont Proxmox",
    "proxmox_clone_parefeu_interface": "Pare-feu interface Proxmox",
    "proxmox_clone_demarrer": "Demarrer le clone apres creation",
}


def charger_yaml(path: Path) -> dict:
    if not path.exists():
        return {}
    with path.open("r", encoding="utf-8") as fichier:
        data = yaml.safe_load(fichier) or {}
    if not isinstance(data, dict):
        raise ValueError(f"{path} ne contient pas une table YAML.")
    return data


def ecrire_yaml(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as fichier:
        fichier.write("---\n")
        fichier.write("# Parametres non sensibles pour les operations Proxmox.\n")
        fichier.write("# Les secrets API doivent aller dans proxmox.vault.yml.\n\n")
        yaml.safe_dump(data, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True)


def demander(identifiant: str, courant: object) -> object:
    libelle = LIBELLES[identifiant]
    if isinstance(courant, bool):
        defaut = "oui" if courant else "non"
        reponse = input(f"{libelle} [{defaut}] : ").strip().lower()
        if not reponse:
            return courant
        return reponse in {"o", "oui", "y", "yes", "true", "1"}

    reponse = input(f"{libelle} [{courant}] : ").strip()
    if not reponse:
        return courant

    if isinstance(courant, int):
        return int(reponse)
    return reponse


def configurer_proxmox() -> None:
    courant = VALEURS_DEFAUT | charger_yaml(FICHIER_PROXMOX)
    nouveau = {}

    print("Configuration Proxmox non sensible")
    print("Entrer pour conserver la valeur entre crochets.\n")
    print("Les valeurs Cloud-Init deja portees par le modele ne sont pas redemandees.\n")

    for identifiant in VALEURS_DEFAUT:
        nouveau[identifiant] = demander(identifiant, courant.get(identifiant, VALEURS_DEFAUT[identifiant]))

    ecrire_yaml(FICHIER_PROXMOX, nouveau)
    print(f"\nEcrit: {os.path.relpath(FICHIER_PROXMOX, RACINE)}")


def demander_oui_non(question: str, defaut: bool = False) -> bool:
    suffixe = "O/n" if defaut else "o/N"
    reponse = input(f"{question} [{suffixe}] : ").strip().lower()
    if not reponse:
        return defaut
    return reponse in {"o", "oui", "y", "yes", "true", "1"}


def configurer_vault() -> None:
    if not demander_oui_non("Configurer les secrets API Proxmox maintenant"):
        return

    if FICHIER_VAULT.exists():
        print(f"Ouverture du Vault existant: {os.path.relpath(FICHIER_VAULT, RACINE)}")
        subprocess.run(["ansible-vault", "edit", str(FICHIER_VAULT)], check=True)
        return

    token_id = input("Token ID Proxmox [set-ops] : ").strip() or "set-ops"
    token_secret = getpass("Token secret Proxmox : ").strip()
    if not token_secret:
        print("Secret vide: Vault non cree.")
        return

    FICHIER_VAULT.parent.mkdir(parents=True, exist_ok=True)
    try:
        fd = os.open(FICHIER_VAULT, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as fichier:
            yaml.safe_dump(
                {
                    "proxmox_api_token_id": token_id,
                    "proxmox_api_token_secret": token_secret,
                },
                fichier,
                default_flow_style=False,
                sort_keys=False,
            )
        subprocess.run(["ansible-vault", "encrypt", str(FICHIER_VAULT)], check=True)
        print(f"Vault cree: {os.path.relpath(FICHIER_VAULT, RACINE)}")
    except Exception:
        if FICHIER_VAULT.exists():
            FICHIER_VAULT.unlink()
        raise


def main() -> int:
    try:
        configurer_proxmox()
        configurer_vault()
    except KeyboardInterrupt:
        print("\nConfiguration interrompue.", file=sys.stderr)
        return 130
    except Exception as exc:
        print(f"erreur: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())