Set-OPS-Public/scripts/site_intrants.py

#!/usr/bin/env python3
"""Les intrants qu'un SITE expose a ses locataires — derives, jamais ecrits.

UN SITE EST UNE INTERFACE, PAS UN DECOR. Un locataire qui veut l'habiter a besoin d'un
petit nombre de valeurs : ou resoudre, ou prendre ses paquets, ou cloner le genome, ou
deposer son etat, quel plan d'administration laisser entrer. Aujourd'hui il les ECRIT A
LA MAIN dans ses intrants, et le depot s'en remet a une garde pour verifier qu'elles
n'ont pas vieilli.

CE QUE CA COUTE, MESURE DEUX FOIS EN DEUX JOURS :
  - `serveur_ops_forge_amont: https://10.0.33.11` chez Chezlepro — la forge avait demenage
    en 10.37.33.11 avec le renumerotage du site. Le commentaire juste au-dessus expliquait
    encore pourquoi l'ancienne adresse figurait dans les SAN du certificat : le raisonnement
    etait intact, la valeur non.
  - `OPS-Technolibre/underlay.yml` decrivait un site d'index 23 quand le vrai disait 31.

Les deux avaient la meme forme : une copie qui a cesse de designer ce qu'elle nommait, et
que rien ne relisait. Une valeur DERIVEE ne peut pas avoir ce defaut.

CE SCRIPT NE MODIFIE RIEN. Il lit le site monte (`underlay.yml` et son `plan/`) et rend le
contrat. `--verifier` le confronte a ce que le locataire monte declare, et c'est la que le
retard se voit.

Usage :
  python3 scripts/site_intrants.py            # le contrat, lisible
  python3 scripts/site_intrants.py --json     # le meme, pour l'API et les gabarits
  python3 scripts/site_intrants.py --verifier # garde : le locataire suit-il son site ?
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))

import underlay as underlay_mod  # noqa: E402
from inventory_rules import dossier_inventaire, instance_courante  # noqa: E402
import yaml  # noqa: E402

PORT_CACHE = 3142
PREFIXE_BINAIRES = "setops-binaires"


def _une(groupe: str) -> str:
    """L'adresse du site qui porte ce groupe, ou '' s'il n'y en a pas.

    PLUSIEURS ADRESSES = AUCUNE REPONSE. Un service porte par deux machines demande un
    choix (lequel amorce ?) que ce script n'a pas a faire silencieusement.
    """
    a = underlay_mod.adresses_site_portant(groupe)
    return a[0] if len(a) == 1 else ""


def _nom_expose(groupe: str, domaine: str) -> str:
    """Le nom de SERVICE que le site publie pour ce groupe, si son plan en declare un."""
    apps = underlay_mod.lire_plan_site("applications.yml").get("applications") or {}
    for a in apps.values():
        if a.get("groupe") == groupe:
            for e in (a.get("expose") or []):
                return str(e)
    hote = next((m["nom"] for m in underlay_mod.machines_site()
                 if groupe in (underlay_mod.services_site().get(m["nom"]) or [])), "")
    return f"{hote}.{domaine}" if hote and domaine else ""


def _opnsense() -> dict:
    """La carte de la frontiere du site (`opnsense.yml`), ou {}."""
    try:
        chemin = underlay_mod.chemin()
        f = chemin.resolve().parent / "opnsense.yml" if chemin else None
        if f and f.is_file():
            return yaml.safe_load(f.read_text(encoding="utf-8")) or {}
    except Exception:
        pass
    return {}


def _ip_publique_site() -> str:
    """L'adresse publique de la frontiere du site, lue dans sa carte (`opnsense.yml`)."""
    return str(_opnsense().get("opnsense_wan_ip") or "")


LIBRE = "libre"


def pool_ips_publiques(carte: dict | None = None) -> dict[str, str]:
    """{adresse : locataire | 'libre'} — le pool que le site attribue (`opnsense.yml`)."""
    carte = _opnsense() if carte is None else carte
    return {str(k): str(v) for k, v in (carte.get("opnsense_ips_publiques") or {}).items()}


def valider_pool(pool: dict[str, str], wan: str, locataires: list[str] | None) -> list[str]:
    """Les fautes du pool : adresse invalide, adresse du site, locataire inconnu ou servi
    deux fois. Une adresse attribuee a deux locataires est impossible par construction
    (la cle est l'adresse) ; un locataire qui en recoit deux, lui, ne l'est pas."""
    import ipaddress
    fautes: list[str] = []
    vus: dict[str, str] = {}
    for ip, qui in sorted(pool.items()):
        try:
            a = ipaddress.ip_address(ip)
        except ValueError:
            fautes.append(f"`{ip}` n'est pas une adresse IP")
            continue
        if a.is_private or a.is_loopback:
            fautes.append(f"`{ip}` n'est pas une adresse publique")
        if ip == wan:
            fautes.append(f"`{ip}` est l'adresse du site (opnsense_wan_ip) : elle n'est pas a attribuer")
        if qui == LIBRE:
            continue
        if locataires is not None and qui not in locataires:
            fautes.append(f"`{ip}` attribuee a « {qui} », que le site n'heberge pas "
                          f"(underlay : {', '.join(locataires) or 'aucun'})")
        if qui in vus:
            fautes.append(f"« {qui} » recoit deux adresses ({vus[qui]}, {ip}) : une par locataire")
        vus[qui] = ip
    return fautes


def ip_publique_de(locataire: str, pool: dict[str, str] | None = None) -> str:
    """L'adresse publique que le site attribue a ce locataire, ou ''."""
    pool = pool_ips_publiques() if pool is None else pool
    return next((ip for ip, qui in sorted(pool.items()) if qui == locataire), "")


def contrat() -> dict:
    u = underlay_mod.charger()
    if not u:
        return {}
    intrants = underlay_mod.lire_plan_site("10-intrants.yml")
    domaine = str(intrants.get("domaine_interne") or "")
    gestion = next((r.get("sous_reseau") for r in underlay_mod.reseaux(u)
                    if r.get("nom") == "management"), "")
    transit = underlay_mod.reseau_transit(u) or {}
    cache = _une("serveur_artefacts")
    forge = _une("serveur_forgejo")

    return {
        # QUI EST CE SITE
        "domaine_site": domaine,
        "index_site": underlay_mod.index_site(u),
        # OU RESOUDRE, AVANT QUE LE LOCATAIRE AIT SON PROPRE RESOLVEUR
        "dns_amorcage": _une("serveur_resolveur"),
        # OU PRENDRE SES PAQUETS, ET SES BINAIRES DE PUBLICATION
        "artefacts_amorcage": f"{cache}:{PORT_CACHE}" if cache else "",
        "setops_depot_binaires": (f"http://{cache}:{PORT_CACHE}/{PREFIXE_BINAIRES}"
                                  if cache else ""),
        # OU CLONER LE GENOME, ET AVEC QUELLE RACINE LUI FAIRE CONFIANCE
        "serveur_ops_forge_amont": f"https://{forge}" if forge else "",
        "forge_amont_ac": str((underlay_mod.chemin() or Path()).resolve().parent
                              / "ac-racine-site.crt"),
        # OU DEPOSER SON ETAT — par NOM : le depot sert plusieurs locataires et peut bouger
        "client_backup_cible": _nom_expose("serveur_backup_site", domaine),
        # QUI TIRE SES ZONES PUBLIQUES — le seul serveur a qui le primaire notifie, et le
        # seul que son pare-feu laisse frapper. Une ADRESSE : c'est un pair de flux.
        "dns_public_site": _une("serveur_dns_public"),
        # VERS OU POINTENT SES NOMS PUBLICS — l'adresse publique de la frontiere. Un fait
        # du site : le locataire la recoit, il ne la recopie pas.
        "ip_publique_site": _ip_publique_site(),
        # L'ADRESSE PUBLIQUE DE CE LOCATAIRE — tiree du pool du site, attribuee par lui
        # (`opnsense_ips_publiques`). Celle de son web frontal et de son courriel.
        "ip_publique": ip_publique_de(instance_courante().name),
        # SOUS QUEL NOM LE SITE FAIT AUTORITE — les NS et le SOA de chaque zone publique.
        "dns_public_nom": str(intrants.get("dns_public_nom") or ""),
        # QUI A LE DROIT D'ENTRER EN SSH
        # QUI A LE DROIT D'ENTRER EN SSH : le plan d'administration du site ET le tunnel
        # WireGuard des admins, derive de sa declaration (jamais recopie).
        "nftables_admin_ssh": ([gestion] if gestion else [])
        + ([str((intrants.get("acces_admin_vpn") or {}).get("reseau"))]
           if (intrants.get("acces_admin_vpn") or {}).get("reseau") else []),
        # PAR OU SORTIR, ET A QUI DEMANDER L'HEURE
        "passerelle_sortie": str(transit.get("passerelle_sortie") or ""),
    }


# Ce que le locataire monte declare, et sous quel nom, pour chaque clef du contrat.
# Vide = le locataire ne declare rien de tel, et c'est legitime (il derive deja).
OU_LE_LOCATAIRE_LE_DIT = {
    "dns_amorcage": ("all/10-intrants.yml", "dns_amorcage"),
    "artefacts_amorcage": ("all/10-intrants.yml", "artefacts_amorcage"),
    "setops_depot_binaires": ("all/10-intrants.yml", "setops_depot_binaires"),
    "serveur_ops_forge_amont": ("serveur_ops.yml", "serveur_ops_forge_amont"),
    "client_backup_cible": ("client_backup.yml", "client_backup_cible"),
    "dns_public_site": ("all/10-intrants.yml", "dns_public_site"),
    "ip_publique_site": ("all/10-intrants.yml", "ip_publique_site"),
    "ip_publique": ("all/10-intrants.yml", "ip_publique"),
    "dns_public_nom": ("all/10-intrants.yml", "dns_public_nom"),
    # CES DEUX-LA MANQUAIENT, ET LE SILENCE A COUTE (2026-09-13). Le site en expose SEPT ;
    # cette table n'en comparait que CINQ. Le controle rapportait donc « CONFORME » avec
    # un aplomb complet, sur un locataire dont le plan d'administration pointait encore
    # les bouts de WAN d'un AUTRE site.
    #
    # CE QUE CA A COUTE : quatorze machines materialisees, le runner insemine — et
    # l'exploitant incapable d'entrer sur son propre runner pour l'armer. `Connection
    # timed out`, sans qu'aucune garde n'ait rien eu a dire.
    #
    # Une liste qui en suit une autre prend du retard : la regle est connue ici, et c'est
    # sa cinquieme occurrence. La garde s'ecrit EN MEME TEMPS que la seconde liste.
    "nftables_admin_ssh": ("all/10-intrants.yml", "nftables_admin_ssh"),
    "passerelle_sortie": ("all/10-intrants.yml", "passerelle_sortie"),
}


def _normaliser(v) -> str:
    """Une forme comparable, que la valeur soit une chaine ou une liste.

    `nftables_admin_ssh` est une LISTE de CIDR ; les cinq intrants d'origine etaient tous
    des chaines, et `str()` suffisait. Comparer `str(['a','b'])` a `str(['b','a'])` ferait
    crier une difference qui n'en est pas une — l'ordre d'une liste de sources ne porte
    aucun sens ici.
    """
    if isinstance(v, (list, tuple)):
        return " ".join(sorted(str(x).strip().rstrip("/") for x in v))
    return str(v).strip().rstrip("/")


def _declare_par_le_locataire() -> dict[str, str]:
    """Lit, sans Ansible, ce que le locataire monte ecrit a propos de son site."""
    # LA RESOLUTION D'INSTANCE EST PARTAGEE, PAS RECOPIEE (P41). La premiere version
    # ecrivait `RACINE / "instance" / "inventories" / "principal"` — le symlink et le nom
    # du dossier en dur. Neuf modules avaient deja porte chacun leur copie, et cinq
    # defauts en etaient sortis en cinq jours. La preuve a mordu sur la dixieme.
    base = dossier_inventaire() / "group_vars"
    vu: dict[str, str] = {}
    for clef, (fichier, variable) in OU_LE_LOCATAIRE_LE_DIT.items():
        f = base / fichier
        if not f.is_file():
            continue
        d = yaml.safe_load(f.read_text(encoding="utf-8")) or {}
        if variable in d:
            vu[clef] = d[variable]
    return vu


def verifier() -> int:
    c = contrat()
    if not c:
        print("Aucun site monte : rien a exposer.")
        return 0
    vu = _declare_par_le_locataire()
    ecarts, absents = [], []
    # LE POOL D'ABORD : une attribution fautive rendrait fausse la valeur meme a comparer.
    ecarts += [f"pool d'adresses publiques du site : {f}" for f in valider_pool(
        pool_ips_publiques(), _ip_publique_site(),
        underlay_mod.tenants_du_site(underlay_mod.charger()))]
    for clef, attendu in c.items():
        if clef not in OU_LE_LOCATAIRE_LE_DIT or not attendu:
            continue
        fichier, variable = OU_LE_LOCATAIRE_LE_DIT[clef]
        if clef not in vu:
            absents.append(f"{variable} ({fichier})")
            continue
        # `setops_depot_binaires` se DERIVE legitimement d'`artefacts_amorcage` en Jinja :
        # une expression n'est pas un retard, c'est le contraire.
        if "{{" in str(vu[clef]):
            continue
        if _normaliser(vu[clef]) != _normaliser(attendu):
            ecarts.append(f"{variable} : le locataire dit « {vu[clef]} », "
                          f"le site expose « {attendu} »")
    if ecarts:
        print("NON CONFORME : le locataire ne suit plus son site.")
        for e in ecarts:
            print("  -", e)
        return 1
    note = f" ({len(absents)} non declare(s), donc derive(s) ou non utilise(s))" if absents else ""
    print(f"CONFORME : {len(vu)} intrant(s) du locataire concordent avec ce que le site expose{note}.")
    return 0


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--verifier", action="store_true")
    args = ap.parse_args(argv)

    if args.verifier:
        return verifier()
    c = contrat()
    if not c:
        print("Aucun site monte (`underlay.yml` absent ou brise).")
        return 0
    if args.json:
        print(json.dumps(c, indent=2, ensure_ascii=False))
        return 0
    print(f"# Intrants exposes par le site « {c['domaine_site'] or '?'} » (index {c['index_site']})")
    print("#")
    print("# A recopier dans les intrants d'un locataire qui veut l'habiter — ou mieux :")
    print("# a comparer aux siens par `--verifier`, qui dit lesquels ont pris du retard.")
    for clef, valeur in c.items():
        if clef in ("domaine_site", "index_site", "forge_amont_ac"):
            continue
        v = json.dumps(valeur, ensure_ascii=False) if isinstance(valeur, list) else valeur
        print(f"{clef}: {v}" if valeur else f"# {clef}: (le site ne porte pas ce service)")
    print(f"#\n# racine de confiance de la forge : {c['forge_amont_ac']}")
    return 0


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