Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity
Set-OPS-Public/scripts/tests/test_temoins.py
Daniel Allaire 11b5bb5733 temoins : l'instantane d'avant rasage est garde, et l'etat remis lui est compare 
M4 a reconstruit Technolibre par le site qui la nomme ; les 7 jeux sont revenus
de l'instantane de 11:34, mais le premier depot d'apres reconstruction l'a chasse
par --keep-daily le jour meme : plus rien a quoi comparer.

Le depot d'avant rasage est etiquete avant-raser et garde jusqu'a la
reconstruction suivante. Le bilan dit ce qui a ete restaure et si c'est encore
au depot. setops-restaurer temoins et make temoins-etat comparent l'etat vivant
a cet instantane : une perte ou une identite changee est un ecart.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-07 12:48:00 -04:00

232 lines
10 KiB
Python
Raw Blame History

#!/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 keycloak 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
\.
"""
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", "<?php $CONFIG = array ('instanceid' => '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", "<?php $CONFIG = array ('instanceid' => '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", "<?php $CONFIG = array ('instanceid' => '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"]
def cluster(bases=("postgres", "keycloak"), roles=("postgres", "keycloak"), comptes=None):
comptes = comptes if comptes is not None else {"public.realm": "2", "public.user_session": "1"}
def interroger(base: str, sql: str):
if "pg_database" in sql:
return list(bases)
if "pg_roles" in sql:
return list(roles)
table = sql.rsplit(" ", 1)[-1]
return [comptes[table]] if table in comptes else None
return interroger
rc, out = lancer(argv, cluster())
verifier(rc == 0 and "postgresql IDENTIQUE 2 table(s)" in out, f"memes bases, roles et lignes : IDENTIQUE ({out.strip()[-120:]})")
rc, out = lancer(argv, cluster(comptes={"public.realm": "2", "public.user_session": "7"}))
verifier(rc == 0 and "user_session : 1 -> 7" in out, "des sessions ouvertes depuis : dit, sans ecart")
rc, out = lancer(argv, cluster(bases=("postgres",)))
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(comptes={"public.user_session": "1"}))
verifier(rc == 1 and "table perdue : keycloak.public.realm" in out, "une table perdue est un ECART")
def main() -> int:
fichiers()
annuaire()
postgresql()
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())
Reference in a new issue View git blame Copy permalink