#!/usr/bin/env python3 """`setops-temoins` : une perte ou une identité changée est un ÉCART ; la vie ordinaire non. POURQUOI (2026-10-07). Les témoins d'une reconstruction se relevaient à la main ; après M4, il ne restait même plus d'instantané d'avant à quoi comparer. Le comparateur qui les remplace doit juger comme l'exploitant jugeait : la racine de l'AC, la clé DKIM, l'`instanceid` de Nextcloud et le mot de passe de `sysadmin` ne bougent JAMAIS ; un courriel lu (`new/` -> `cur/`, drapeau `:2,S`) n'est pas un courriel perdu ; une table de sessions qui grossit n'est pas une base perdue. Sur des arbres FABRIQUÉS (un « avant » extrait, un « vivant »), chaque règle est éprouvée dans les deux sens : ce qui doit passer passe, ce qui doit échouer échoue. """ from __future__ import annotations import base64 import contextlib import io import shutil import sys import tempfile from pathlib import Path RACINE = Path(__file__).resolve().parents[2] sys.path.insert(0, str(RACINE / "roles/client_backup/files")) import importlib.util # noqa: E402 _spec = importlib.util.spec_from_file_location("setops_temoins", RACINE / "roles/client_backup/files/setops-temoins.py") T = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(T) ECHECS: list[str] = [] def verifier(cond: bool, msg: str) -> None: print(("OK " if cond else "ECHEC ") + msg) if not cond: ECHECS.append(msg) def ecrire(p: Path, texte: str) -> None: p.parent.mkdir(parents=True, exist_ok=True) p.write_text(texte) LDIF = """dn: dc=exemple,dc=internal objectClass: domain dc: exemple entryCSN: 20261007153000.000000Z#000000#000#000000 dn: uid=sysadmin,ou=people,dc=exemple,dc=internal objectClass: inetOrgPerson uid: sysadmin cn: Sysadmin userPassword:: {mdp} modifyTimestamp: 20261001000000Z dn: uid=marie,ou=people,dc=exemple,dc=internal objectClass: inetOrgPerson uid: marie cn: Marie mail: marie@exemple.internal """ DUMP = r"""CREATE ROLE keycloak; CREATE ROLE postgres; CREATE DATABASE template1 WITH TEMPLATE = template0 ENCODING = 'UTF8'; CREATE DATABASE keycloak WITH TEMPLATE = template0 ENCODING = 'UTF8'; CREATE DATABASE icingadb WITH TEMPLATE = template0 ENCODING = 'UTF8'; \connect keycloak COPY public.realm (id, name) FROM stdin; 1 maitre 2 exemple \. COPY public.user_session (id) FROM stdin; a \. \connect icingadb COPY public.host (id, name) FROM stdin; \\x0a mon-01 \. COPY public.history (id, event_time) FROM stdin; \\x01 1 \\x02 2 \. """ def ldif(mdp: str = "e1NTSEF9YW5jaWVu", marie: bool = True, mail: str = "marie@exemple.internal") -> str: t = LDIF.replace("{mdp}", mdp).replace("marie@exemple.internal", mail) if not marie: t = t.split("dn: uid=marie")[0] return t def lancer(argv: list[str], interroger=None) -> tuple[int, str]: sortie = io.StringIO() with contextlib.redirect_stdout(sortie): rc = T.main(argv, interroger=interroger or T.psql) return rc, sortie.getvalue() def fichiers() -> None: with tempfile.TemporaryDirectory() as d: d = Path(d) avant, vivant = d / "avant", d / "vivant" # L'AC : une cle, un certificat, et sa base qui se compacte d'elle-meme. for racine in (avant / str(vivant).lstrip("/"), vivant): ecrire(racine / "step-ca/certs/root_ca.crt", "RACINE") ecrire(racine / "step-ca/secrets/root_ca_key", "CLE") ecrire(racine / "vmail/marie/Maildir/cur/1700.M1.h,S=10:2,S", "bonjour") ecrire(racine / "rspamd/dkim/setops.key", "DKIM") ecrire(racine / "nc/config/config.php", " 'oc1abc',);") ecrire(racine / "nc/data/marie/files/rapport.odt", "rapport") ecrire(avant / str(vivant).lstrip("/") / "step-ca/db/000001.vlog", "ancien journal") ecrire(vivant / "step-ca/db/000002.vlog", "journal compacte") ecrire(avant / str(vivant).lstrip("/") / "vmail/marie/Maildir/new/1800.M2.h,S=5", "salut") ecrire(vivant / "vmail/marie/Maildir/cur/1800.M2.h,S=5:2,S", "salut") # lu depuis ecrire(vivant / "vmail/marie/Maildir/new/1900.M3.h,S=4", "neuf") # recu depuis ecrire(avant / str(vivant).lstrip("/") / "rspamd/bayes.spam.sqlite", "appris") ecrire(vivant / "rspamd/bayes.spam.sqlite", "appris davantage") ecrire(avant / str(vivant).lstrip("/") / "nc/data/appdata_oc1abc/preview/1.png", "cache") jeux = [f"step_ca={vivant}/step-ca", f"courriel={vivant}/vmail", f"rspamd={vivant}/rspamd", f"nextcloud={vivant}/nc/data:{vivant}/nc/config"] base = ["--avant", str(avant), "--instantane", "abcd1234", "--hote", "h"] argv = base + [x for j in jeux for x in ("--jeu", j)] rc, out = lancer(argv) verifier(rc == 0, f"la vie ordinaire n'est pas un ecart : base de l'AC compactee, courriel lu et " f"recu, bayes appris, cache Nextcloud refait (code {rc})\n{out}") verifier("step_ca CONFORME" in out and "rspamd CONFORME" in out, "des changements ordinaires sont dits CONFORME, pas IDENTIQUE") verifier("perdu : msg" not in out, "un courriel passe de new/ a cur/ avec un drapeau n'est pas perdu") ecrire(vivant / "step-ca/secrets/root_ca_key", "AUTRE CLE") rc, out = lancer(argv) verifier(rc == 1 and "IDENTITE secrets/root_ca_key" in out, "une cle de l'AC changee est un ECART") ecrire(vivant / "step-ca/secrets/root_ca_key", "CLE") ecrire(vivant / "rspamd/dkim/setops.key", "AUTRE DKIM") rc, out = lancer(argv) verifier(rc == 1 and "IDENTITE dkim/setops.key" in out, "une cle DKIM changee est un ECART") ecrire(vivant / "rspamd/dkim/setops.key", "DKIM") ecrire(vivant / "nc/config/config.php", " 'ocNEUF',);") rc, out = lancer(argv) verifier(rc == 1 and "instanceid : oc1abc -> ocNEUF" in out, "un instanceid Nextcloud change est un ECART") ecrire(vivant / "nc/config/config.php", " 'oc1abc',);") (vivant / "nc/data/marie/files/rapport.odt").unlink() rc, out = lancer(argv) verifier(rc == 1 and "perdu : marie/files/rapport.odt" in out, "un fichier d'une personne perdu est un ECART") ecrire(vivant / "nc/data/marie/files/rapport.odt", "rapport") shutil.rmtree(vivant / "vmail/marie/Maildir/cur") rc, out = lancer(argv) verifier(rc == 1 and "perdu : marie/Maildir/msg/1700.M1.h,S=10" in out, "un courriel perdu est un ECART") rc, out = lancer(base) verifier(rc == 0 and "SANS OBJET" in out, "un noeud sans etat n'a rien a comparer") def annuaire() -> None: with tempfile.TemporaryDirectory() as d: d = Path(d) ecrire(d / "avant/var/backups/setops/openldap/annuaire.ldif", ldif()) vivant = d / "vivant.ldif" argv = ["--avant", str(d / "avant"), "--instantane", "x", "--jeu", "openldap=/var/backups/setops/openldap", "--ldif", "/var/backups/setops/openldap/annuaire.ldif", "--ldif-vivant", str(vivant)] # slapadd repose les attributs operationnels, et une ligne pliee n'est pas une autre valeur. ecrire(vivant, ldif().replace("modifyTimestamp: 20261001000000Z", "modifyTimestamp: 20261007160000Z") .replace("mail: marie@exemple.internal", "mail: marie@exemple.inter\n nal")) rc, out = lancer(argv) verifier(rc == 0 and "openldap IDENTIQUE" in out, f"attributs operationnels et pliage ignores : IDENTIQUE ({out.strip()[-160:]})") ecrire(vivant, ldif(mdp=base64.b64encode(b"{SSHA}amorcage").decode())) rc, out = lancer(argv) verifier(rc == 1 and "IDENTITE userPassword : uid=sysadmin" in out, "le mot de passe de sysadmin remis a l'amorcage est un ECART (la panne du 2026-09-30)") ecrire(vivant, ldif(marie=False)) rc, out = lancer(argv) verifier(rc == 1 and "perdue : uid=marie" in out, "une entree perdue est un ECART") ecrire(vivant, ldif(mail="marie@ailleurs.internal")) rc, out = lancer(argv) verifier(rc == 0 and "CONFORME" in out and "(mail)" in out, "un attribut ordinaire modifie est dit, sans ecart") def postgresql() -> None: with tempfile.TemporaryDirectory() as d: d = Path(d) ecrire(d / "avant/var/backups/setops/postgresql/toutes-bases.sql", DUMP) argv = ["--avant", str(d / "avant"), "--instantane", "x", "--jeu", "postgresql=/var/backups/setops/postgresql", "--dump", "/var/backups/setops/postgresql/toutes-bases.sql"] AVANT = {"public.realm": ["1", "2"], "public.user_session": ["a"], "public.host": ["\\\\x0a"], "public.history": ["\\\\x01", "\\\\x02"]} def cluster(bases=("postgres", "template0", "template1", "keycloak", "icingadb"), roles=("postgres", "keycloak"), **tables): contenu = {**AVANT, **tables} def interroger(base: str, sql: str): if "pg_database" in sql: # Comme PostgreSQL : `template1` est un modele, qu'un filtre sur # `datistemplate` ecarterait (le faux « base perdue » du premier essai reel). return [b for b in bases if not ("datistemplate" in sql and b == "template1")] if "pg_roles" in sql: return list(roles) table = sql.split(" from ", 1)[1].split(")", 1)[0] return list(contenu[table]) if contenu.get(table) is not None else None return interroger rc, out = lancer(argv, cluster()) verifier(rc == 0 and "postgresql IDENTIQUE 4 table(s)" in out and "template0" not in out, f"memes bases, roles et cles : IDENTIQUE ({out.strip()[-120:]})") rc, out = lancer(argv, cluster(**{"public.user_session": ["b", "c"]})) verifier(rc == 0 and "public.user_session : 1 retiree(s), 2 nouvelle(s)" in out, "des sessions renouvelees : dit, sans ecart (la vie ordinaire de Keycloak)") rc, out = lancer(argv, cluster(**{"public.history": ["\\\\x01", "\\\\x02", "\\\\x03"]})) verifier(rc == 0 and "1 nouvelle(s)" in out, "un historique qui s'allonge : dit, sans ecart") rc, out = lancer(argv, cluster(**{"public.history": ["\\\\x02", "\\\\x03"]})) verifier(rc == 1 and "historique perdu : icingadb.public.history : 1 ligne(s)" in out, "une ligne d'historique perdue est un ECART") # Le cas reel du 2026-10-07 : meme nombre de lignes, aucune en commun. Compter ne le # voyait pas ; c'etait la base d'Icinga, non restauree. rc, out = lancer(argv, cluster(**{"public.host": ["\\\\xff"], "public.history": ["\\\\x08", "\\\\x09"]})) verifier(rc == 1 and "base icingadb : aucune des 3 ligne(s) d'avant ne subsiste" in out, "une base entierement renouvelee, a nombre de lignes egal, est un ECART") rc, out = lancer(argv, cluster(bases=("postgres", "template1", "icingadb"))) verifier(rc == 1 and "base perdue : keycloak" in out, "une base perdue est un ECART") rc, out = lancer(argv, cluster(roles=("postgres",))) verifier(rc == 1 and "role perdu : keycloak" in out, "un role perdu est un ECART") rc, out = lancer(argv, cluster(**{"public.realm": None})) verifier(rc == 1 and "table perdue : keycloak.public.realm" in out, "une table perdue est un ECART") def icinga() -> None: with tempfile.TemporaryDirectory() as d: d = Path(d) vivant = d / "vivant" for racine in (d / "avant" / str(vivant).lstrip("/"), vivant): ecrire(racine / "icinga2/ca/ca.crt", "AC") ecrire(racine / "icinga2/ca/ca.key", "CLE") ecrire(racine / "icinga2/icingadb.env", '"37503b53fd8"') ecrire(vivant / "icinga2/ca/serial.txt", "02") argv = ["--avant", str(d / "avant"), "--instantane", "x", "--jeu", f"icinga={vivant}/icinga2/ca:{vivant}/icinga2/icingadb.env"] rc, out = lancer(argv) verifier(rc == 0, f"l'AC d'Icinga remise, un certificat de noeud signe depuis : sans ecart ({out.strip()[-120:]})") ecrire(vivant / "icinga2/icingadb.env", '"0afe46c3"') rc, out = lancer(argv) verifier(rc == 1 and "IDENTITE icingadb.env" in out, "un environnement d'Icinga DB change est un ECART (l'historique deviendrait orphelin)") ecrire(vivant / "icinga2/icingadb.env", '"37503b53fd8"') ecrire(vivant / "icinga2/ca/ca.key", "AUTRE CLE") rc, out = lancer(argv) verifier(rc == 1 and "IDENTITE ca.key" in out, "une cle d'AC d'Icinga changee est un ECART") def main() -> int: fichiers() annuaire() postgresql() icinga() if ECHECS: print(f"\n{len(ECHECS)} echec(s).") return 1 print("\nLes temoins voient une perte ou une identite changee, et laissent passer la vie ordinaire.") return 0 if __name__ == "__main__": sys.exit(main())