#!/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())