Set-OPS-Public/scripts/forge_cles.py

#!/usr/bin/env python3
"""Le site enregistre, sur le depot de chaque locataire, la cle qui le lira. Rien d'autre.

POURQUOI (2026-10-11, `docs/depots-prives-conception.md`). Les depots des locataires deviennent
prives sur la forge du site. Chaque runner de locataire lit LE SIEN avec une cle de deploiement
EN LECTURE SEULE : la cle publique que sa face reseau publie (`genome.cle_lecture`). Le site
l'enregistre avec ses propres identifiants d'administration de la forge ; aucun secret ne passe
du locataire au site, ni l'inverse.

CE QUE LE SCRIPT TOUCHE, EXACTEMENT. Les seules cles de deploiement intitulees
`lecture — runner de <locataire>`, sur le seul depot que la face nomme :
  - absente           -> ajoutee, en lecture seule ;
  - perimee (autre cle sous le meme titre) -> retiree : une cle qui n'est plus publiee est une
                         revocation, la garder ouvrirait le depot a un runner disparu ;
  - presente en ECRITURE -> refus : une cle de lecture qui pourrait pousser n'est pas a corriger
                         en silence.
La cle du runner du SITE (`runner du site — ...`, posee par `genome_pousser.yml`) n'est jamais
touchee. Seuls les locataires dont la face nomme CE site comme hebergeur sont traites.

Lance sur le runner du site (il porte les faces et joint la forge), par `make forge-cles`.
Le mot de passe d'administration arrive par l'environnement (`FORGEJO_ADMIN`), depuis la voute
du site ; il n'est jamais affiche.

Usage :
  FORGEJO_ADMIN=... python3 scripts/forge_cles.py              # montre ce qui serait fait
  FORGEJO_ADMIN=... python3 scripts/forge_cles.py --confirmer
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path

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

PREFIXE_TITRE = "lecture — runner de "


def titre_de(locataire: str) -> str:
    return f"{PREFIXE_TITRE}{locataire}"


def _deux(cle: str) -> str:
    """Le type et le corps d'une cle publique : le commentaire ne l'identifie pas."""
    return " ".join(str(cle or "").split()[:2])


def demandes(site_nom: str | None = None) -> list[dict]:
    """(locataire, depot, cle) de chaque locataire dont la face nomme ce site hebergeur."""
    import contexte as _ctx
    sortie = []
    for site, loc in _ctx.couples_hebergeur():
        if site_nom and site.nom != site_nom:
            continue
        face = loc.face_publiee() or {}
        if face.get("hebergeur") != site.nom:
            continue
        g = face.get("genome") or {}
        sortie.append({"locataire": loc.nom, "depot": str(g.get("depot") or ""),
                       "cle": _deux(g.get("cle_lecture"))})
    return sortie


def plan(voulues: list[dict], existantes: dict[str, list[dict]]) -> tuple[list, list, list]:
    """Ce qu'il faut ajouter, retirer, et refuser. Pur : aucun appel a la forge.

    `existantes` : depot -> cles de deploiement telles que l'API les rend
    ({id, title, key, read_only}).
    """
    ajouter, retirer, refus = [], [], []
    for v in voulues:
        if not v["depot"] or not v["cle"]:
            refus.append(f"{v['locataire']} : sa face ne publie ni depot ni cle de lecture")
            continue
        cles = existantes.get(v["depot"], [])
        titre = titre_de(v["locataire"])
        meme_cle = [k for k in cles if _deux(k.get("key")) == v["cle"]]
        if any(not k.get("read_only", True) for k in meme_cle):
            refus.append(f"{v['locataire']} : sa cle de lecture est enregistree EN ECRITURE sur "
                         f"{v['depot']} — a retirer a la main, apres verification")
            continue
        if not meme_cle:
            ajouter.append(v)
        for k in cles:
            if k.get("title") == titre and _deux(k.get("key")) != v["cle"]:
                retirer.append({**v, "id": k.get("id")})
    return ajouter, retirer, refus


class Forge:
    def __init__(self, base: str, organisation: str, mot_de_passe: str,
                 utilisateur: str = "forgejo-admin"):
        self.base = base.rstrip("/")
        self.org = organisation
        self.auth = base64.b64encode(f"{utilisateur}:{mot_de_passe}".encode()).decode()
        self.ctx = ssl.create_default_context()

    def _api(self, chemin: str, methode: str = "GET", corps: dict | None = None):
        req = urllib.request.Request(f"{self.base}/api/v1{chemin}", method=methode,
                                     data=json.dumps(corps).encode() if corps is not None else None)
        req.add_header("Authorization", f"Basic {self.auth}")
        req.add_header("Content-Type", "application/json")
        try:
            with urllib.request.urlopen(req, timeout=20, context=self.ctx) as r:
                texte = r.read().decode() or "null"
                return r.status, json.loads(texte)
        except urllib.error.HTTPError as e:
            return e.code, (e.read().decode(errors="replace") or "")[:200]

    def cles(self, depot: str) -> list[dict]:
        st, rep = self._api(f"/repos/{self.org}/{depot}/keys?limit=50")
        if st != 200:
            raise RuntimeError(f"lecture des cles de {depot} : HTTP {st} {rep}")
        return rep or []

    def ajouter(self, depot: str, titre: str, cle: str) -> int:
        return self._api(f"/repos/{self.org}/{depot}/keys", "POST",
                         {"title": titre, "key": cle, "read_only": True})[0]

    def retirer(self, depot: str, ident: int) -> int:
        return self._api(f"/repos/{self.org}/{depot}/keys/{ident}", "DELETE")[0]


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--confirmer", action="store_true")
    ap.add_argument("--forge", default=os.environ.get("SETOPS_FORGE_URL", ""),
                    help="https://forge.<domaine> (defaut : derive du plan du site)")
    ap.add_argument("--organisation", default="genome")
    ap.add_argument("--site", help="ne traiter que les locataires de ce site")
    a = ap.parse_args(argv)
    mdp = os.environ.get("FORGEJO_ADMIN", "")
    if not mdp:
        print("REFUS : FORGEJO_ADMIN absent (mot de passe d'administration de la forge, voute du site).",
              file=sys.stderr)
        return 2
    base = a.forge
    if not base:
        import underlay as underlay_mod
        dom = str((underlay_mod.lire_plan_site("10-intrants.yml") or {}).get("domaine_interne") or "")
        if not dom:
            print("REFUS : ni --forge ni `domaine_interne` au plan du site.", file=sys.stderr)
            return 2
        base = f"https://forge.{dom}"
    voulues = demandes(a.site)
    if not voulues:
        print("Aucun locataire ne nomme ce site hebergeur : rien a enregistrer.")
        return 0
    forge = Forge(base, a.organisation, mdp)
    existantes = {v["depot"]: forge.cles(v["depot"]) for v in voulues if v["depot"]}
    ajouter, retirer, refus = plan(voulues, existantes)
    for v in voulues:
        etat = ("a ajouter" if v in ajouter else "refus" if any(v["locataire"] in r for r in refus)
                else "deja enregistree")
        print(f"  {v['locataire']:<18} {v['depot']:<18} {v['cle'][:32]}…  {etat}")
    for x in retirer:
        print(f"  {x['locataire']:<18} {x['depot']:<18} cle perimee #{x['id']}  a retirer")
    for r in refus:
        print(f"  REFUS {r}")
    if not a.confirmer:
        print("\nRien n'a ete ecrit. Relancer avec --confirmer (make forge-cles CONFIRMER=true).")
        return 1 if refus else 0
    echecs = 0
    for x in retirer:
        st = forge.retirer(x["depot"], x["id"])
        echecs += st not in (204, 404)
        print(f"  retiree  {x['depot']} #{x['id']} : HTTP {st}")
    for v in ajouter:
        st = forge.ajouter(v["depot"], titre_de(v["locataire"]), v["cle"])
        echecs += st not in (201,)
        print(f"  ajoutee  {v['depot']} ({v['locataire']}) : HTTP {st}")
    # RELIRE, PAS SE FIER AU CODE DE RETOUR (D-68).
    relu = {v["depot"]: forge.cles(v["depot"]) for v in voulues if v["depot"]}
    reste_a, reste_r, _ = plan(voulues, relu)
    if reste_a or reste_r or echecs or refus:
        print(f"\nECART apres ecriture : {len(reste_a)} a ajouter, {len(reste_r)} a retirer, "
              f"{echecs} echec(s), {len(refus)} refus.")
        return 1
    print("\nChaque depot de locataire porte exactement la cle de lecture que sa face publie.")
    return 0


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