Set-OPS-Public/scripts/devis_certificats.py

#!/usr/bin/env python3
"""Devis des certificats : ce que le disque porte contre ce que la mémoire sert.

Lit les relevés déposés par `playbooks/maintenance/devis-certificats.yml`. Aucun accès
réseau ici. Sort en code 1 s'il y a un écart.

Le défaut traqué est connu et récurrent (voir `docs/autorisation.md` et le journal) : un
certificat renouvelé SUR DISQUE mais toujours servi PÉRIMÉ depuis la mémoire du service,
tant que nginx / postfix / dovecot / slapd n'ont pas été rechargés. Rien ne le signale —
le fichier est bon, `step` est content, et le client voit un certificat expiré.

**Ce que ce devis a appris en se construisant.** Les certificats step-ca vivent 24 h et le
minuteur les renouvelle toutes les ~14 min. Une empreinte servie DIFFÉRENTE de l'empreinte
sur disque est donc l'état NORMAL, pas un défaut : comparer les empreintes aurait produit
un vérificateur qui crie en permanence. Le signal utile est ailleurs :

  1. l'échéance de ce qui est RÉELLEMENT SERVI — c'est ça qui tombe ;
  2. un hôte qui sert un certificat sans déclarer aucun service à recharger : rien ne
     rafraîchira jamais sa copie en mémoire.
"""

from __future__ import annotations

import glob
import json
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

RACINE = Path(__file__).resolve().parent.parent
MOTIF = str(RACINE / "instance" / "devis-certificats.json.*")

# Un certificat servi qui expire dans moins que ça n'a plus de marge : le service doit
# etre recharge avant. Genereux par rapport au renouvellement (~14 min), serre par
# rapport a la duree de vie (24 h).
SEUIL_HEURES = 6


def _champ(texte: str, cle: str) -> str:
    m = re.search(rf"^{re.escape(cle)}=(.*)$", texte, re.M)
    return m.group(1).strip() if m else ""


def _echeance(valeur: str) -> datetime | None:
    try:
        return datetime.strptime(valeur, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=timezone.utc)
    except ValueError:
        return None


def _servis(texte: str) -> list[dict]:
    blocs, courant = [], None
    for ligne in texte.splitlines():
        if ligne.startswith("PORT="):
            courant = {"port": ligne[5:], "brut": ""}
            blocs.append(courant)
        elif courant is not None:
            courant["brut"] += ligne + "\n"
    for b in blocs:
        b["subject"] = _champ(b["brut"], "subject")
        b["fin"] = _echeance(_champ(b["brut"], "notAfter"))
        m = re.search(r"Fingerprint=([0-9A-F:]+)", b["brut"])
        b["empreinte"] = m.group(1) if m else ""
    return blocs


def analyser(releves: list[dict]) -> tuple[list[str], list[str]]:
    ecarts, notes = [], []
    maintenant = datetime.now(timezone.utc)

    for r in sorted(releves, key=lambda x: x["hote"]):
        hote, recharges = r["hote"], r.get("recharges_declares") or []

        if r.get("disque_rc", 1) != 0 or not r.get("disque"):
            ecarts.append(f"{hote}: aucun certificat lisible en {r.get('chemin', '?')}")
            continue

        fin_disque = _echeance(_champ(r["disque"], "notAfter"))
        emp_disque = (re.search(r"Fingerprint=([0-9A-F:]+)", r["disque"]) or [None, ""])[1]
        if fin_disque and (fin_disque - maintenant).total_seconds() < SEUIL_HEURES * 3600:
            reste = (fin_disque - maintenant).total_seconds() / 3600
            ecarts.append(f"{hote}: le certificat SUR DISQUE expire dans {reste:.1f} h "
                          f"— le renouvellement automatique ne fait plus son travail")

        for s in _servis(r["servis"]):
            if not s["fin"]:
                continue
            reste = (s["fin"] - maintenant).total_seconds() / 3600
            perime = s["empreinte"] != emp_disque
            if reste < SEUIL_HEURES:
                ecarts.append(
                    f"{hote}:{s['port']} sert un certificat qui expire dans {reste:.1f} h"
                    + (f" — recharger {', '.join(recharges)}" if recharges
                       else " — et AUCUN service a recharger n'est declare"))
            elif perime and not recharges:
                notes.append(
                    f"{hote}:{s['port']} sert une copie plus ancienne que le disque, et "
                    f"aucun `client_pki_reload_services` n'est declare : rien ne la "
                    f"rafraichira jamais")
    return ecarts, notes


def main(argv: list[str] | None = None) -> int:
    fichiers = sorted(glob.glob(argv[0] if argv else MOTIF))
    if not fichiers:
        print("Relevés absents. Lancer d'abord : make certificats-plan", file=sys.stderr)
        return 2

    releves = [json.loads(Path(f).read_text(encoding="utf-8")) for f in fichiers]
    ecarts, notes = analyser(releves)
    print(f"Devis des certificats — {len(releves)} hôte(s), seuil {SEUIL_HEURES} h\n")

    for n in notes:
        print(f"  [latent]  {n}")
    if notes and ecarts:
        print()
    for e in ecarts:
        print(f"  [écart]   {e}")

    if not ecarts:
        print(("\n" if notes else "") + "CONFORME : aucun certificat servi n'est en fin de vie.")
        return 0
    print(f"\nÉCART : {len(ecarts)} certificat(s) en difficulté.")
    return 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))