Set-OPS-Public/scripts/genome_colis.py

#!/usr/bin/env python3
"""Porter des commits jusqu'a une forge qu'on ne route pas — par colis, pas par reseau.

POURQUOI CE DETOUR, ET POURQUOI IL EST BON.

La forge du genome vit DANS le site, sur un reseau que le poste de l'exploitant ne route
pas — et c'est voulu : le poste n'a de patte que sur l'administration, la frontiere police
le reste. Il ne peut donc pas pousser lui-meme.

On ne perce pas un chemin pour lui. Le RUNNER existe exactement pour ca : il est dans le
site, il porte le genome, il a une identite. On lui PORTE les commits dans un `git bundle`
— un fichier, verifiable, qui ne demande aucun reseau — et c'est LUI qui pousse.

Mesure du 2026-08-26 : la forge du site etait restee quatre commits derriere le poste, sans
que rien ne le signale. Le runner clonait donc un moteur perime — dont, ce jour-la, le
correctif qui desarme le pare-feu Proxmox. Un ecosysteme qui se reproduit depuis une forge
en retard reproduit ses defauts.

DEUX GESTES, DEUX COTES :

    genome_colis.py emballer <depot> <colis>
        Sur le CONTROLEUR. Fabrique le colis a partir de `main`.

    genome_colis.py recevoir <depot> <colis> <pushurl> [--cle <fichier>]
        Sur le RUNNER. Verifie le colis, avance `main` EN AVANCE RAPIDE SEULEMENT,
        puis pousse. Rend du JSON : {avant, apres, pousse}.

    genome_colis.py etat <depot>
        Des deux cotes. Ce que cette copie porte, pour le confronter a la forge.

`--ff-only` est le garde-fou. Si les deux histoires ont diverge, on s'arrete plutot que de
fabriquer une fusion que personne n'a decidee — et le message de git nomme alors exactement
ce qu'il faut reconcilier.

CE SCRIPT VIT DANS LE MOTEUR, donc le runner en a une copie des qu'il l'a clone : on ne
depose rien de special sur la machine pour qu'elle sache recevoir.
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path


def _git(depot: Path, *args: str, env: dict | None = None) -> str:
    """Un git, dans un depot, qui parle clair quand il echoue."""
    r = subprocess.run(["git", "-C", str(depot), *args],
                       capture_output=True, text=True, env=env)
    if r.returncode != 0:
        sortie = (r.stderr or r.stdout or "").strip()
        raise SystemExit(f"git {' '.join(args)} a echoue dans {depot} :\n  {sortie}")
    return (r.stdout or "").strip()


def emballer(depot: Path, colis: Path, branche: str) -> int:
    if not (depot / ".git").exists():
        raise SystemExit(f"{depot} n'est pas un depot git — rien a emballer.")
    colis.parent.mkdir(parents=True, exist_ok=True)
    _git(depot, "bundle", "create", str(colis), branche)
    print(json.dumps({"depot": str(depot), "colis": str(colis),
                      "tete": _git(depot, "log", "-1", "--format=%H")}))
    return 0


def etat(depot: Path, branche: str) -> int:
    """Ce que cette copie porte — pour le confronter a ce que la forge porte.

    LA DERIVE NE SE VOIT PAS TOUTE SEULE. Le 2026-08-26, la forge du site etait restee
    quatre commits derriere le poste sans que rien ne le signale ; le runner clonait donc
    un moteur perime, dont le correctif qui desarme le pare-feu Proxmox. On rend donc
    l'ecart LISIBLE, plutot que d'esperer s'en souvenir.
    """
    if not (depot / ".git").exists():
        raise SystemExit(f"{depot} n'est pas un depot git.")
    print(json.dumps({
        "tete": _git(depot, "log", "-1", "--format=%H", branche),
        "courte": _git(depot, "log", "-1", "--format=%h", branche),
        "sujet": _git(depot, "log", "-1", "--format=%s", branche)[:80],
        "propre": _git(depot, "status", "--porcelain") == "",
    }))
    return 0


def recevoir(depot: Path, colis: Path, pushurl: str, cle: str | None,
             branche: str) -> int:
    if not (depot / ".git").exists():
        raise SystemExit(f"{depot} n'est pas un depot git — rien a recevoir.")
    if not colis.is_file():
        raise SystemExit(f"{colis} est introuvable : le colis n'est pas arrive.")

    # UN COLIS DOIT ETRE VERIFIABLE AVANT D'ETRE CRU. `git bundle verify` exige d'etre
    # lance DEPUIS un depot — il verifie que les prerequis du colis sont deja dans cette
    # histoire-ci. Lance ailleurs : « la verification d'un colis requiert un depot ».
    _git(depot, "bundle", "verify", str(colis))

    avant = _git(depot, "log", "-1", "--format=%H")
    _git(depot, "fetch", str(colis),
         f"refs/heads/{branche}:refs/remotes/colis/{branche}")
    _git(depot, "merge", "--ff-only", f"refs/remotes/colis/{branche}")
    apres = _git(depot, "log", "-1", "--format=%H")

    env = None
    if cle:
        import os
        env = dict(os.environ)
        env["GIT_SSH_COMMAND"] = (
            f"ssh -o StrictHostKeyChecking=accept-new -o BatchMode=yes -i {cle}")

    # `-c remote.origin.pushurl=…` plutot qu'un remote ecrit : la LECTURE reste anonyme
    # (doctrine de `serveur_ops` — aucun justificatif a detenir pour se reconstruire),
    # seule l'ECRITURE s'authentifie, et seulement le temps de cette commande.
    r = subprocess.run(
        ["git", "-C", str(depot), "-c", f"remote.origin.pushurl={pushurl}",
         "push", "origin", branche],
        capture_output=True, text=True, env=env)
    sortie = ((r.stdout or "") + (r.stderr or "")).strip()
    if r.returncode != 0:
        raise SystemExit(f"la poussee a echoue :\n  {sortie}")

    print(json.dumps({"avant": avant, "apres": apres,
                      "pousse": f"{branche} -> {branche}" in sortie, "sortie": sortie[:400]}))
    return 0


def main(argv: list[str]) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    sous = p.add_subparsers(dest="geste", required=True)

    e = sous.add_parser("emballer", help="fabriquer un colis depuis `main` (controleur)")
    e.add_argument("depot")
    e.add_argument("colis")
    e.add_argument("--branche", default="main")

    r = sous.add_parser("recevoir", help="verifier, avancer, pousser (runner)")
    r.add_argument("depot")
    r.add_argument("colis")
    r.add_argument("pushurl")
    r.add_argument("--cle", default=None, help="cle SSH du runner")
    r.add_argument("--branche", default="main")

    d = sous.add_parser("etat", help="ce que cette copie porte")
    d.add_argument("depot")
    d.add_argument("--branche", default="main")

    a = p.parse_args(argv)
    if a.geste == "etat":
        return etat(Path(a.depot), a.branche)
    if a.geste == "emballer":
        return emballer(Path(a.depot), Path(a.colis), a.branche)
    return recevoir(Path(a.depot), Path(a.colis), a.pushurl, a.cle, a.branche)


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