Set-OPS-Public/scripts/forge_amorcer.py

#!/usr/bin/env python3
"""Amorce la forge d'un SITE neuf avec le genome. Le maillon qui n'existait pas.

POURQUOI CE SCRIPT EXISTE (2026-09-12, premiere reconstruction du site depuis zero).

Le runner du site clone le genome depuis SA PROPRE forge — `serveur_ops_forge_hote`
derive de `groups['serveur_forgejo']`. L'ordre des couches est juste : `serveur_forgejo`
precede `serveur_ops`. Mais `roles/serveur_forgejo` ne cree NI organisation NI depot : a
froid, la forge naît vide, et le `git clone` du runner echoue apres trois essais.

LE DEPLOIEMENT N'AVAIT JAMAIS RENCONTRE CE CAS parce que la forge existait depuis le
premier jour, remplie a la main, a une date que personne n'a notee. C'est exactement ce
qu'une limite « jamais reconstruit depuis zero » cache : pas une fragilite, un MAILLON
QUI N'AVAIT JAMAIS ETE FABRIQUE PAR LE CODE.

POURQUOI DEPUIS LE POSTE, ET PAS DEPUIS UN ROLE. A cet instant precis, le poste est le
SEUL endroit qui detienne le genome : la forge est vide et le runner n'existe pas encore.
Un role qui tenterait de se servir lui-meme tournerait en rond. C'est un geste d'amorcage,
et un amorcage vient toujours de l'exterieur de ce qu'il amorce.

LA LISTE VIENT DE LA MEME DECLARATION QUE LE RUNNER LIT — `serveur_ops_depots` dans le
plan du site. Ecrire une seconde liste ici la ferait diverger de celle qui compte, et ce
depot passe son temps a supprimer ce genre de paire.

Usage :
  python3 scripts/forge_amorcer.py                 # montre ce qui serait pousse
  python3 scripts/forge_amorcer.py --confirmer
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from urllib.parse import urlsplit
from pathlib import Path

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

import underlay as underlay_mod  # noqa: E402


def _depots_declares() -> list[dict]:
    """`serveur_ops_depots` du plan du SITE — la source que le runner lit."""
    import yaml
    plan = underlay_mod.lire_plan_site("serveurs.yml") or {}
    for srv in (plan.get("serveurs") or {}).values():
        d = ((srv.get("variables") or {}).get("serveur_ops_depots"))
        if d:
            return list(d)
    return []


def _api(base: str, chemin: str, methode: str = "GET", corps: dict | None = None,
         auth: str = "") -> tuple[int, object]:
    req = urllib.request.Request(f"{base}/api/v1{chemin}", method=methode)
    req.add_header("Authorization", f"Basic {auth}")
    req.add_header("Content-Type", "application/json")
    donnees = json.dumps(corps).encode() if corps is not None else None
    import ssl
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE   # la racine step-ca n'est pas encore au magasin du poste
    try:
        with urllib.request.urlopen(req, donnees, timeout=25, context=ctx) as r:
            return r.status, json.loads(r.read() or b"null")
    except urllib.error.HTTPError as e:
        return e.code, (e.read() or b"").decode()[:200]
    except OSError as e:
        return 0, str(e)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--forge", help="URL de la forge (defaut : derivee du plan du site)")
    ap.add_argument("--organisation", default="genome")
    ap.add_argument("--utilisateur", default="forgejo-admin")
    ap.add_argument("--cle-publique",
                    default=str(Path.home() / ".ssh" / "id_ed25519_ansible_chezlepro.pub"),
                    help="cle publique a enregistrer dans la forge neuve (pour le push SSH)")
    ap.add_argument("--confirmer", action="store_true")
    a = ap.parse_args()

    depots = _depots_declares()
    if not depots:
        print("Aucun `serveur_ops_depots` au plan du site : rien a amorcer.")
        return 2

    # L'hote de la forge vient du plan, comme pour le runner.
    import yaml
    apps = (underlay_mod.lire_plan_site("applications.yml") or {}).get("applications") or {}
    hote = next((str(x.get("hote")) for x in apps.values()
                 if str(x.get("groupe")) == "serveur_forgejo"), "")
    fqdn = next((str(f) for x in apps.values()
                 if str(x.get("groupe")) == "serveur_forgejo"
                 for f in (x.get("expose") or [])), "")
    base = a.forge or (f"https://{fqdn}" if fqdn else "")
    if not base:
        print("La forge du site ne se derive pas du plan : passer --forge.")
        return 2

    print(f"Forge      : {base}   (machine {hote or '?'})")
    print(f"Organisation : {a.organisation}\n")

    manquants, prets = [], []
    for d in depots:
        local = FRERES / str(d.get("dest"))
        if not (local / ".git").is_dir():
            manquants.append(f"{d['dest']} — absent du poste ({local})")
        else:
            prets.append((str(d["depot"]), local, str(d.get("branche") or "main")))

    for nom, local, branche in prets:
        print(f"  {nom:22} <- {local.name:20} branche {branche}")
    for m in manquants:
        print(f"  MANQUANT  {m}")

    if manquants:
        print(f"\nREFUS : {len(manquants)} depot(s) declare(s) n'existent pas sur ce poste.\n"
              f"Amorcer une forge avec une liste incomplete la rendrait credible et fausse :\n"
              f"le runner clonerait ce qui est la et echouerait sur le reste, plus tard.")
        return 2

    if not a.confirmer:
        print(f"\n{len(prets)} depot(s) seraient crees puis pousses.\n"
              f"Relancer avec :  make forge-amorcer CONFIRMER=true")
        return 0

    # PURGER L'ENTREE PERIMEE DE L'HOTE QU'ON S'APPRETE A CONTACTER (2026-09-12).
    #
    # `accept-new` accepte une PREMIERE rencontre, jamais un CHANGEMENT — et c'est bien
    # ainsi. Mais une forge reconstruite porte une cle d'hote inedite sur une adresse
    # DEJA CONNUE : ssh refuse, `git` ne peut rien demander sans terminal, et le message
    # qu'il rend parle de droits d'acces et de depot inexistant. Mesure au tour 2 : un
    # aller-retour de diagnostic complet pour une cle d'hote.
    #
    # CE QUE CA N'AFFAIBLIT PAS. La purge ne vaut que pour l'hote de CETTE forge, au
    # moment ou l'exploitant amorce une machine qu'il vient de creer lui-meme par une API
    # authentifiee. On ne touche a aucune autre entree, et `accept-new` reprend ensuite
    # son role : la cle notee ici sera defendue comme les autres au prochain contact.
    # LA CIBLE EST L'HOTE QUE `git` VA CONTACTER, pas celui de l'API : quand l'amorcage
    # passe par un tunnel, `--forge` vaut `127.0.0.1` alors que le push part vers le nom
    # du remote. Purger le mauvais ne servirait a rien, et on ne s'en apercevrait qu'au
    # push.
    hotes_git = set()
    for _nom, local, _br in prets:
        url = subprocess.run(["git", "-C", str(local), "remote", "get-url", "origin"],
                             capture_output=True, text=True, timeout=20).stdout.strip()
        h = urlsplit(url).hostname if "://" in url else (
            url.split("@", 1)[-1].split(":", 1)[0] if "@" in url else "")
        if h and h not in ("127.0.0.1", "localhost"):
            hotes_git.add(h)
    for cible in sorted(hotes_git):
        r = subprocess.run(["ssh-keygen", "-R", cible], capture_output=True, text=True, timeout=20)
        if r.returncode == 0 and "not found" not in (r.stderr or "").lower():
            print(f"  known_hosts : entree perimee de {cible} retiree")

    mdp = os.environ.get("SETOPS_FORGE_MDP", "")
    if not mdp:
        print("\nREFUS : mot de passe de l'administrateur de la forge absent.\n"
              "  export SETOPS_FORGE_MDP=\"$(ansible-vault view … | …)\"  puis relancer.\n"
              "Il n'est pas lu du depot : un secret ne traverse pas un argument de ligne\n"
              "de commande, ou il finirait dans l'historique du shell.")
        return 2
    auth = base64.b64encode(f"{a.utilisateur}:{mdp}".encode()).decode()

    # LA FORGE NEUVE NE CONNAIT PERSONNE (2026-09-12). Les remotes `origin` poussent en
    # `ssh://git@…` ; une Forgejo fraiche rejette la cle de l'exploitant — `Permission
    # denied (publickey)` — parce qu'aucune cle n'y est enregistree. Amorcer sans cela
    # revient a creer des depots qu'on ne peut pas remplir.
    cle = Path(a.cle_publique)
    if cle.is_file():
        st, rep = _api(base, "/user/keys", "POST",
                       {"title": f"amorcage-{cle.stem}", "key": cle.read_text().strip()}, auth)
        if st in (201, 200):
            print(f"  cle publique {cle.name} : enregistree")
        elif st == 422:
            print(f"  cle publique {cle.name} : deja la")
        else:
            print(f"  cle publique {cle.name} : ECHEC {st} {rep}")
            return 2
    else:
        print(f"  cle publique absente ({cle}) — le push SSH echouera")
        return 2

    st, _ = _api(base, f"/orgs/{a.organisation}", auth=auth)
    if st == 404:
        st, rep = _api(base, "/orgs", "POST", {"username": a.organisation}, auth)
        print(f"  organisation {a.organisation} : "
              + ("creee" if st in (201, 200) else f"ECHEC {st} {rep}"))
    elif st == 200:
        print(f"  organisation {a.organisation} : deja la")
    else:
        print(f"  organisation {a.organisation} : ECHEC {st}")
        return 2

    echecs = 0
    for nom, local, branche in prets:
        st, _ = _api(base, f"/repos/{a.organisation}/{nom}", auth=auth)
        if st == 404:
            st, rep = _api(base, f"/orgs/{a.organisation}/repos", "POST",
                           {"name": nom, "private": False}, auth)
            # PUBLIC DANS LE PERIMETRE DE LA FORGE, ET C'EST VOULU (2026-09-12).
            #
            # Cree en `private` au premier essai — par reflexe. Le runner clone
            # ANONYMEMENT : la tache `git` de `serveur_ops` ne porte aucun identifiant,
            # et l'echec ne dit pas « depot prive », il dit :
            #
            #     could not read Username for 'https://forge.genese.internal'
            #
            # La forge n'est joignable que depuis les zones du site, les supernets
            # locataires et le plan d'administration : « public » y signifie « lisible
            # par ceux qui ont deja franchi la frontiere ». Les secrets, eux, sont dans
            # des voutes chiffrees — c'est la qu'est la protection, pas dans un drapeau
            # de visibilite qui casserait la reproduction.
            if st not in (201, 200):
                print(f"  {nom:22} ECHEC creation {st} {rep}")
                echecs += 1
                continue
        # `--force` JAMAIS : si le depot porte deja quelque chose, on veut le savoir,
        # pas l'ecraser. Un amorcage n'a de sens que sur une forge vide.
        # LA FORGE EST NEUVE, SA CLE D'HOTE AUSSI (2026-09-12). Une reconstruction donne
        # a chaque machine une cle d'hote inedite : `known_hosts` porte l'ancienne, `git`
        # refuse — a juste titre — et sans terminal il ne peut rien demander :
        #
        #     Host key verification failed.
        #     ssh_askpass: exec(/usr/bin/ssh-askpass): No such file or directory
        #
        # `accept-new` ACCEPTE UNE PREMIERE RENCONTRE, JAMAIS UN CHANGEMENT : si une cle
        # connue differe, ssh refuse toujours. C'est la posture juste ici — l'exploitant
        # vient de creer cette machine lui-meme, il y a quelques minutes, par une API
        # authentifiee. Exiger qu'il ait deja vu la cle d'un hote qui n'existait pas
        # rendrait l'amorcage impossible.
        #
        # NE PAS confondre avec `StrictHostKeyChecking=no`, qui avalerait aussi un
        # changement — c'est-a-dire exactement l'attaque contre laquelle il protege.
        env = dict(os.environ, GIT_SSH_COMMAND="ssh -o StrictHostKeyChecking=accept-new")
        r = subprocess.run(["git", "-C", str(local), "push", "origin", branche],
                           capture_output=True, text=True, timeout=180, env=env)
        if r.returncode == 0:
            print(f"  {nom:22} pousse ({branche})")
        else:
            # LA DERNIERE LIGNE DE `stderr` EST LA PLUS GENERIQUE (2026-09-12). `git`
            # termine par « assurez-vous que le depot existe » quelle que soit la cause :
            # cle d'hote changee, droits, depot absent. N'afficher qu'elle a coute un
            # aller-retour de diagnostic — on rend les premieres lignes, qui portent le
            # motif reel.
            _err = [l for l in (r.stderr or "").strip().splitlines() if l.strip()]
            print(f"  {nom:22} ECHEC push :")
            for _l in _err[:4]:
                print(f"      {_l[:110]}")
            echecs += 1

    print(f"\n{len(prets) - echecs}/{len(prets)} depot(s) amorces.")
    return 1 if echecs else 0


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