Daniel Allaire
82a2c4a4c2
L'epreuve du colis sur le runner de Technolibre : git bundle verify exige un depot existant, et l'insemination est le cas ou il n'existe pas encore. Le clone verifie le colis lui-meme ; un depot existant le verifie avant de tirer. Les colis restes d'un passage interrompu sont retires. make forge-cles rend aussi chaque depot prive ou public selon le role que lui donne le plan du site, et seulement une fois les cles de lecture en place. 95 preuves. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
241 lines
11 KiB
Python
241 lines
11 KiB
Python
#!/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 "
|
|
# LA VISIBILITE SUIT LE ROLE (2026-10-11) : le plan d'un ecosysteme et la carte du site sont
|
|
# prives ; le moteur et les modeles, publics. La MEME liste que `serveur_ops_roles_prives`.
|
|
ROLES_PRIVES = {"instance", "tenant", "hebergeur"}
|
|
|
|
|
|
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
|
|
|
|
|
|
def visibilites_voulues() -> dict[str, bool]:
|
|
"""depot -> prive ?, d'apres `serveur_ops_depots` du plan du site monte (son runner)."""
|
|
import underlay as underlay_mod
|
|
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 {str(x["depot"]): str(x.get("role")) in ROLES_PRIVES for x in d}
|
|
return {}
|
|
|
|
|
|
def plan_visibilite(voulues: dict[str, bool], actuelles: dict[str, bool]) -> list[tuple[str, bool]]:
|
|
"""(depot, prive voulu) pour chaque depot dont la visibilite differe. Pur."""
|
|
return [(d, v) for d, v in sorted(voulues.items()) if d in actuelles and actuelles[d] != v]
|
|
|
|
|
|
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 prive(self, depot: str) -> bool | None:
|
|
st, rep = self._api(f"/repos/{self.org}/{depot}")
|
|
return bool(rep.get("private")) if st == 200 and isinstance(rep, dict) else None
|
|
|
|
def rendre(self, depot: str, prive: bool) -> int:
|
|
return self._api(f"/repos/{self.org}/{depot}", "PATCH", {"private": prive})[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)
|
|
forge = Forge(base, a.organisation, mdp)
|
|
# LES CLES D'ABORD, LA VISIBILITE ENSUITE : un depot ne devient prive que quand son runner
|
|
# peut deja le lire. L'ordre inverse couperait le runner entre les deux.
|
|
vis_voulues = visibilites_voulues()
|
|
vis_actuelles = {d: v for d in vis_voulues if (v := forge.prive(d)) is not None}
|
|
a_basculer = plan_visibilite(vis_voulues, vis_actuelles)
|
|
if not voulues and not a_basculer:
|
|
print("Aucun locataire ne nomme ce site hebergeur, et chaque depot a sa visibilite.")
|
|
return 0
|
|
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")
|
|
# LA FIN DE LA CLE, PAS SON DEBUT : toutes les ed25519 commencent par les memes 32
|
|
# caracteres, et deux lignes differentes s'affichaient identiques (2026-10-11).
|
|
print(f" {v['locataire']:<18} {v['depot']:<18} …{v['cle'][-16:]} {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}")
|
|
print("Visibilite des depots (d'apres le role que leur donne le plan du site) :")
|
|
for d, v in sorted(vis_voulues.items()):
|
|
act = vis_actuelles.get(d)
|
|
etat = ("introuvable" if act is None else "ok" if act == v
|
|
else f"a rendre {'prive' if v else 'public'}")
|
|
print(f" {d:<18} {'prive ' if v else 'public'} {etat}")
|
|
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}")
|
|
# LA VISIBILITE, SEULEMENT SI LES CLES SONT EN PLACE (voir plus haut).
|
|
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 des cles : {len(reste_a)} a ajouter, {len(reste_r)} a retirer, "
|
|
f"{echecs} echec(s), {len(refus)} refus. Aucune visibilite n'a ete changee.")
|
|
return 1
|
|
for d, v in a_basculer:
|
|
st = forge.rendre(d, v)
|
|
echecs += st != 200
|
|
print(f" rendu {'prive ' if v else 'public'} {d} : HTTP {st}")
|
|
# RELIRE, PAS SE FIER AU CODE DE RETOUR (D-68).
|
|
vis_relues = {d: forge.prive(d) for d in vis_voulues}
|
|
reste_v = plan_visibilite(vis_voulues, {d: x for d, x in vis_relues.items() if x is not None})
|
|
if reste_v or echecs:
|
|
print(f"\nECART apres ecriture : {len(reste_v)} visibilite(s) a corriger, {echecs} echec(s).")
|
|
return 1
|
|
print("\nChaque depot de locataire porte exactement la cle de lecture que sa face publie, "
|
|
"et chaque depot la visibilite que son role demande.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|