90 lines
3.8 KiB
Python
90 lines
3.8 KiB
Python
|
#!/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 voute(base: Path) -> dict:
|
||
|
|
"""Secrets, lus par ansible-vault. Rien n'est journalise ni ecrit sur disque."""
|
||
|
|
for nom in ("principal", "production", "lab"):
|
||
|
|
p = base / "inventories" / nom / "group_vars" / "all" / "vault.yml"
|
||
|
|
if p.is_file():
|
||
|
|
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 {}
|
||
|
|
raise SystemExit(f"Aucune voute sous {base}/inventories/*/group_vars/all/.")
|
||
|
|
|
||
|
|
|
||
|
|
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 "")
|