Daniel Allaire
d2e2bb8990
make flux ecrit, a cote de chaque .nft et depuis les memes regles, la liste de ce que chaque VM doit joindre. La sonde (nftables_baseline) la teste chaque minute avec la cause (rejete, delai, personne n'ecoute) ; client_sante la porte en mode minute ; Icinga a une fraicheur par sonde. eprouver_parefeu --sondes : tout activer d'un coup, juger par Icinga. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
431 lines
22 KiB
Python
431 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Activer le pare-feu Proxmox d'UNE VM de locataire, en prouvant qu'il ne coupe rien.
|
|
|
|
POURQUOI UNE PROCEDURE, ET PAS SEULEMENT `proxmox-fw-appliquer` (2026-09-28). Aucune des 26 VM
|
|
des locataires n'avait son pare-feu actif : les reconstructions clonent sans ses options.
|
|
Tout activer d'un coup ouvrait 26 pannes possibles, et chaque defaut ne s'est montre qu'a
|
|
l'activation d'UNE VM : le ping de supervision qu'aucune regle ne portait (l'ICMP etait saute
|
|
par le devis), le tunnel d'administration du locataire que nftables admettait et Proxmox non.
|
|
Une VM reconstruite perd de nouveau ses options : cette procedure servira encore.
|
|
|
|
CE QU'ELLE FAIT, POUR UNE VM :
|
|
1. MATRICE tiree du DEVIS lui-meme : pour chaque regle des groupes de la VM, CHAQUE membre
|
|
de la source autorisee tente VM:port. On teste ce qui est promis, pas ce qu'on croit ;
|
|
2. OBSERVATION : les connexions entrantes ETABLIES sur la VM (trois releves), confrontees
|
|
aux regles. La matrice ne voit que le DECLARE ; un flux reel non declare la passerait,
|
|
puis serait coupe. Un seul flux hors regles : REFUS d'activer ;
|
|
3. matrice AVANT. Un echec deja present : REFUS — il faut le comprendre d'abord. Sauf si
|
|
rien n'ECOUTE sur ce port hors de 127.0.0.1 : « sans objet » (un nginx lie en local
|
|
derriere une passerelle SSO, un frontal sans site) ;
|
|
4. ACTIVATION par le runner du site (seul a joindre l'API du cluster) : `--vm <vmid>` ;
|
|
5. matrice APRES, comparee ; puis sondes de sante relancees et critiques d'Icinga.
|
|
|
|
python3 scripts/eprouver_parefeu.py --instance OPS-X --hote idm-01 --plan
|
|
python3 scripts/eprouver_parefeu.py --instance OPS-X --hote idm-01 --activer
|
|
|
|
LIMITE CONNUE : un flux UDP est teste en TCP sur le meme port (la matrice ne sait pas ouvrir
|
|
une « connexion » UDP). Pour un resolveur, verifier une vraie resolution (`dig +notcp`).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
RACINE = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(RACINE / "scripts"))
|
|
|
|
from devis_reseau import DOSSIER_INSTANCES, admin_avec_tunnel # noqa: E402
|
|
|
|
|
|
def _inventaire(instance: str) -> Path:
|
|
return DOSSIER_INSTANCES / instance / "inventories" / "principal" / "hosts.yml"
|
|
|
|
|
|
def _ip_par_hote(instance: str) -> dict[str, str]:
|
|
import yaml
|
|
table: dict[str, str] = {}
|
|
|
|
def w(n):
|
|
if isinstance(n, dict):
|
|
for k, v in n.items():
|
|
if k == "hosts" and isinstance(v, dict):
|
|
for h, hv in v.items():
|
|
if isinstance(hv, dict) and hv.get("ansible_host"):
|
|
table[h] = hv["ansible_host"]
|
|
else:
|
|
w(v)
|
|
w(yaml.safe_load(_inventaire(instance).read_text(encoding="utf-8")))
|
|
return table
|
|
|
|
|
|
def _ansible(instance: str, hotes: str, commande: str) -> dict[str, str]:
|
|
"""{hote: sortie} d'une commande shell jouee en root sur des hotes du locataire."""
|
|
r = subprocess.run(["ansible", "-i", str(_inventaire(instance)), hotes, "-b", "-m", "shell",
|
|
"-a", commande], capture_output=True, text=True, cwd=RACINE)
|
|
sorties, hote = {}, None
|
|
for ligne in r.stdout.splitlines():
|
|
m = re.match(r"^(\S+) \| \w+ \| rc=\d+ >>$", ligne)
|
|
if m:
|
|
hote = m.group(1)
|
|
sorties[hote] = ""
|
|
elif hote:
|
|
sorties[hote] += ligne + "\n"
|
|
return sorties
|
|
|
|
|
|
def matrice(instance: str, hote: str) -> tuple[list[dict], int, dict[str, list]]:
|
|
env = dict(os.environ, SETOPS_INSTANCE=str(DOSSIER_INSTANCES / instance))
|
|
r = subprocess.run([sys.executable, "scripts/devis_proxmox_fw.py", "--json"],
|
|
capture_output=True, text=True, cwd=RACINE, env=env)
|
|
if r.returncode != 0:
|
|
raise SystemExit("Le devis ne se genere pas :\n" + r.stderr[:400])
|
|
devis = json.loads(r.stdout)
|
|
ips = _ip_par_hote(instance)
|
|
h_par_ip = {ip: h for h, ip in ips.items()}
|
|
# LE BLOC DE CE LOCATAIRE, PAS LE PREMIER QUI PORTE CE NOM. Les locataires partagent
|
|
# leurs noms d'hotes (`ops-01` existe chez chacun) : prendre la premiere affectation
|
|
# venue visait la VM d'un autre — constate au premier essai de ce script.
|
|
for b in (x for x in devis["blocs"] if x.get("tenant") == instance):
|
|
a = next((x for x in b["affectations"] if x["hote"] == hote), None)
|
|
if a is None:
|
|
continue
|
|
groupes = {g["nom"]: g for g in b["groupes"]}
|
|
tests = []
|
|
# LES MEMBRES QUI SONT DES RESEAUX, PAS DES MACHINES (2026-09-30). On ne peut pas les
|
|
# TESTER (aucune machine a qui demander d'ouvrir la connexion), mais Proxmox les
|
|
# applique : `t23-admin` porte la zone d'administration du site. Les ignorer faisait
|
|
# prendre le navigateur de l'exploitant, sur l'edge en 443, pour un flux « hors des
|
|
# regles » — et `--flotte` s'arretait. `!reseau` = exclusion (`nomatch`).
|
|
reseaux: dict[str, list] = {}
|
|
for gn in a["groupes"]:
|
|
for regle in groupes[gn]["regles"]:
|
|
membres = b["ipsets"].get(regle["source"].lstrip("+"), {}).get("membres", [])
|
|
port = str(regle.get("dport") or regle.get("icmp_type"))
|
|
pos = [m for m in membres if "/" in str(m) and not str(m).startswith("!")]
|
|
neg = [str(m)[1:] for m in membres if str(m).startswith("!")]
|
|
if pos:
|
|
reseaux.setdefault(port, []).append(
|
|
([ipaddress.ip_network(x, strict=False) for x in pos],
|
|
[ipaddress.ip_network(x, strict=False) for x in neg]))
|
|
for m in membres:
|
|
src = h_par_ip.get(m)
|
|
if src and src != hote:
|
|
tests.append({"src": src, "ip": ips[hote], "proto": regle["proto"],
|
|
"port": str(regle.get("dport") or regle.get("icmp_type")),
|
|
"groupe": gn})
|
|
return tests, int(a["vmid"]), reseaux
|
|
raise SystemExit(f"REFUS : « {hote} » n'a aucune affectation au devis de {instance}.")
|
|
|
|
|
|
def jouer(instance: str, hote: str, tests: list[dict]) -> dict[str, str]:
|
|
par_src: dict[str, list[dict]] = {}
|
|
for t in tests:
|
|
par_src.setdefault(t["src"], []).append(t)
|
|
res: dict[str, str] = {}
|
|
for src, xs in par_src.items():
|
|
lignes = []
|
|
for t in xs:
|
|
cle = f"{t['src']}>{hote}:{t['proto']}/{t['port']}"
|
|
essai = (f"ping -c1 -W2 {t['ip']} >/dev/null 2>&1" if t["proto"] == "icmp"
|
|
else f"timeout 3 bash -c '</dev/tcp/{t['ip']}/{t['port']}' 2>/dev/null")
|
|
lignes.append(f"{essai} && echo '{cle} ok' || echo '{cle} echec'")
|
|
for sortie in _ansible(instance, src, "; ".join(lignes)).values():
|
|
for ligne in sortie.splitlines():
|
|
if ligne.endswith((" ok", " echec")):
|
|
k, v = ligne.rsplit(" ", 1)
|
|
res[k] = v
|
|
return res
|
|
|
|
|
|
def ecoutes(instance: str, hote: str) -> set[str]:
|
|
"""Ports ecoutes HORS de la boucle locale : ce qu'un voisin peut atteindre."""
|
|
sortie = _ansible(instance, hote, "ss -Hltn").get(hote, "")
|
|
ports = set()
|
|
for ligne in sortie.splitlines():
|
|
champs = ligne.split()
|
|
if len(champs) >= 4:
|
|
adr, _, port = champs[3].rpartition(":")
|
|
if not adr.startswith(("127.", "[::1]")):
|
|
ports.add(port)
|
|
return ports
|
|
|
|
|
|
def observer(instance: str, hote: str, tests: list[dict],
|
|
reseaux: dict[str, list] | None = None) -> list[str]:
|
|
"""Les connexions entrantes etablies qu'AUCUNE regle ne couvre."""
|
|
cmd = ('L=" $(ss -Hltn | awk \'{n=split($4,a,":"); print a[n]}\' | sort -u | tr "\\n" " ") "; '
|
|
'for i in 1 2 3; do ss -Hnt state established | awk -v L="$L" '
|
|
'\'{n=split($3,a,":"); lp=a[n]; m=split($4,b,":"); p=b[1]; for(k=2;k<m;k++) p=p":"b[k]; '
|
|
'if (index(L," " lp " ")) print p" "lp}\'; sleep 5; done | sort -u')
|
|
sortie = _ansible(instance, hote, cmd).get(hote, "")
|
|
ips = _ip_par_hote(instance)
|
|
h_par_ip = {ip: h for h, ip in ips.items()}
|
|
permis: dict[str, set[str]] = {}
|
|
for t in tests:
|
|
permis.setdefault(t["port"], set()).add(t["src"])
|
|
admin = [ipaddress.ip_network(n) for n in admin_avec_tunnel(instance)]
|
|
hors = []
|
|
for ligne in sortie.splitlines():
|
|
champs = ligne.split()
|
|
if len(champs) != 2:
|
|
continue
|
|
ip = champs[0].strip("[]").replace("::ffff:", "")
|
|
port = champs[1]
|
|
if ip.startswith("127.") or ip == "::1":
|
|
continue
|
|
src = h_par_ip.get(ip, ip)
|
|
if src == hote:
|
|
continue # vers soi-meme par sa propre adresse : `lo`, pas le pont Proxmox
|
|
if src in permis.get(port, set()):
|
|
continue
|
|
if port == "22" and any(ipaddress.ip_address(ip) in n for n in admin):
|
|
continue
|
|
adr = ipaddress.ip_address(ip)
|
|
if any(any(adr in n for n in pos) and not any(adr in n for n in neg)
|
|
for pos, neg in (reseaux or {}).get(port, [])):
|
|
continue
|
|
hors.append(f"{src} -> {hote}:{port}")
|
|
return sorted(set(hors))
|
|
|
|
|
|
def activer(instance: str, vmid: int, args) -> int:
|
|
distante = (f"cd {args.racine_runner}/Set-OPS-public && export PATH={args.racine_runner}/venv/bin:$PATH "
|
|
f"&& SETOPS_INSTANCE={args.racine_runner}/{instance} make proxmox-fw-appliquer "
|
|
f"CONFIRMER=true ARGS='--vm {vmid}'")
|
|
cmd = ["ssh", "-o", "BatchMode=yes", "-J", args.rebond, args.runner,
|
|
f"sudo -u setops bash -lc {json.dumps(distante)}"]
|
|
r = subprocess.run(cmd, capture_output=True, text=True)
|
|
for ligne in r.stdout.splitlines():
|
|
if re.search(r"~ VM|ECHEC|reconcilie|INCOMPLETE|Rien a faire", ligne):
|
|
print(" ", ligne.strip())
|
|
return 0 if ("reconcilie" in r.stdout or "Rien a faire" in r.stdout) and "ECHEC" not in r.stdout else 1
|
|
|
|
|
|
def critiques_icinga(instance: str) -> list[str]:
|
|
_ansible(instance, "all", "systemctl start setops-sante.service; true")
|
|
cmd = ("P=$(sed -n '/ApiUser \"root\"/,/}/p' /etc/icinga2/conf.d/api-users.conf | "
|
|
"grep -oP 'password = \"\\K[^\"]+'); "
|
|
"curl -sk -u \"root:$P\" -X POST -H 'Accept: application/json' "
|
|
"https://localhost:5665/v1/actions/reschedule-check "
|
|
"-d '{\"type\":\"Service\",\"filter\":\"service.name==\\\"ping4\\\"\",\"force\":true}' >/dev/null; "
|
|
"sleep 20; curl -sk -u \"root:$P\" "
|
|
"'https://localhost:5665/v1/objects/services?filter=service.state==2&attrs=last_check_result'")
|
|
sortie = next(iter(_ansible(instance, "serveur_icinga", cmd).values()), "{}")
|
|
try:
|
|
return [x["name"] for x in json.loads(sortie[sortie.index("{"):])["results"]]
|
|
except (ValueError, KeyError):
|
|
return ["(Icinga illisible)"]
|
|
|
|
|
|
def vmids(instance: str) -> dict[str, int]:
|
|
"""{hote: vmid} du devis de ce locataire — la meme source que l'activation."""
|
|
env = dict(os.environ, SETOPS_INSTANCE=str(DOSSIER_INSTANCES / instance))
|
|
r = subprocess.run([sys.executable, "scripts/devis_proxmox_fw.py", "--json"],
|
|
capture_output=True, text=True, cwd=RACINE, env=env)
|
|
if r.returncode != 0:
|
|
raise SystemExit("Le devis ne se genere pas :\n" + r.stderr[:400])
|
|
table = {}
|
|
for b in (x for x in json.loads(r.stdout)["blocs"] if x.get("tenant") == instance):
|
|
for a in b["affectations"]:
|
|
table[a["hote"]] = int(a["vmid"])
|
|
return table
|
|
|
|
|
|
def etat_sonde(instance: str, nom: str) -> dict[str, tuple[int, str, float]]:
|
|
"""{hote: (etat, texte, fin du dernier rapport)} du service `nom`, lu dans Icinga."""
|
|
cmd = ("P=$(sed -n '/ApiUser \"root\"/,/}/p' /etc/icinga2/conf.d/api-users.conf | "
|
|
"grep -oP 'password = \"\\K[^\"]+'); "
|
|
"curl -sk -u \"root:$P\" -H 'Accept: application/json' "
|
|
f"'https://localhost:5665/v1/objects/services?filter=service.name==%22{nom}%22"
|
|
"&attrs=host_name&attrs=state&attrs=last_check_result'")
|
|
sortie = next(iter(_ansible(instance, "serveur_icinga", cmd).values()), "{}")
|
|
table = {}
|
|
try:
|
|
for x in json.loads(sortie[sortie.index("{"):])["results"]:
|
|
a = x["attrs"]
|
|
r = a.get("last_check_result") or {}
|
|
table[a["host_name"]] = (int(a.get("state", 3)), str(r.get("output", "")),
|
|
float(r.get("execution_end") or 0))
|
|
except (ValueError, KeyError):
|
|
pass
|
|
return table
|
|
|
|
|
|
def activer_par_sondes(instance: str, args) -> int:
|
|
"""La flotte entiere d'un coup, jugee par la sonde `connectivite` (2026-09-30).
|
|
|
|
La matrice VM par VM prenait une vingtaine de minutes en fin de reconstruction, pour une
|
|
preuve valable a l'instant de l'activation seulement. Ici : chaque VM teste deja, chaque
|
|
minute, les flux que le registre lui promet, et le dit a Icinga. On verifie que tout est
|
|
sain AVANT, on active tout, on attend les rapports suivants, et seul ce qui CHANGE compte.
|
|
Perdu, en connaissance de cause : l'isolement d'une faute a une seule VM — la sonde, elle,
|
|
nomme le couple exact qui ne passe plus.
|
|
"""
|
|
table = vmids(instance)
|
|
avant = etat_sonde(instance, "connectivite")
|
|
manquent = sorted(set(table) - set(avant))
|
|
if manquent:
|
|
print(f" REFUS : aucune sonde `connectivite` dans Icinga pour {', '.join(manquent)} — "
|
|
"redeployer `serveur_durci` et `client_sante`, puis `serveur_icinga`.")
|
|
return 2
|
|
deja = {h: v for h, v in avant.items() if h in table and v[0] == 2}
|
|
if deja:
|
|
print(" REFUS : des flux sont DEJA coupes, avant toute activation — les comprendre d'abord :")
|
|
for h, (_, texte, _) in sorted(deja.items()):
|
|
print(f" {h} : {texte[:180]}")
|
|
return 2
|
|
print(f" avant : {len(table)} VM, `connectivite` saine partout "
|
|
f"({sum(1 for h in table if avant[h][0] == 1)} avertissement(s))")
|
|
crit_avant = set(critiques_icinga(instance))
|
|
if args.plan:
|
|
print(" PLAN SEUL — rien n'a ete active.")
|
|
return 0
|
|
t0 = time.time()
|
|
print(f" activation de {len(table)} VM d'un coup a {time.strftime('%H:%M:%S')}")
|
|
distante = (f"cd {args.racine_runner}/Set-OPS-public && export PATH={args.racine_runner}/venv/bin:$PATH "
|
|
f"&& SETOPS_INSTANCE={args.racine_runner}/{instance} make proxmox-fw-appliquer "
|
|
f"CONFIRMER=true ARGS='{' '.join(f'--vm {v}' for v in sorted(table.values()))}'")
|
|
r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-J", args.rebond, args.runner,
|
|
f"sudo -u setops bash -lc {json.dumps(distante)}"], capture_output=True, text=True)
|
|
if not (("reconcilie" in r.stdout or "Rien a faire" in r.stdout) and "ECHEC" not in r.stdout):
|
|
print(" ECHEC de l'activation :")
|
|
for ligne in (r.stdout + r.stderr).splitlines()[-8:]:
|
|
print(" " + ligne)
|
|
return 3
|
|
# On attend que CHAQUE VM ait rapporte APRES l'activation (rapport a la minute, `ttl` 3 min).
|
|
limite = t0 + 300
|
|
while True:
|
|
time.sleep(20)
|
|
apres = etat_sonde(instance, "connectivite")
|
|
frais = [h for h in table if h in apres and apres[h][2] > t0 + 5]
|
|
print(f" {time.strftime('%H:%M:%S')} : {len(frais)}/{len(table)} VM ont rapporte depuis l'activation")
|
|
if len(frais) == len(table) or time.time() > limite:
|
|
break
|
|
absents = sorted(set(table) - set(frais))
|
|
coupes = {h: apres[h] for h in frais if apres[h][0] == 2}
|
|
for h, (_, texte, _) in sorted(coupes.items()):
|
|
print(f" COUPE {h} : {texte[:200]}")
|
|
crit = sorted(set(critiques_icinga(instance)) - crit_avant)
|
|
print(f" Icinga : {len(crit)} critique(s) apparu(s)" + (f" — {', '.join(crit)}" if crit else ""))
|
|
if absents:
|
|
print(f" SANS NOUVELLE de {', '.join(absents)} apres 5 min — muette, ou coupee d'Icinga.")
|
|
if coupes or crit or absents:
|
|
print(" RETOUR ARRIERE d'une VM : `enable=0` sur /nodes/<noeud>/qemu/<vmid>/firewall/options "
|
|
f"(vmid : {', '.join(f'{h}={table[h]}' for h in sorted(set(coupes) | set(absents)))})")
|
|
return 4
|
|
print(f" {len(table)}/{len(table)} VM : pare-feu Proxmox actif, `connectivite` saine partout.")
|
|
return 0
|
|
|
|
|
|
def ordre_flotte(instance: str) -> list[str]:
|
|
"""Toutes les VM du locataire, son RUNNER EN DERNIER.
|
|
|
|
C'est depuis le poste qu'on joue la matrice, mais c'est le runner qui conduit le reste
|
|
de la vie du locataire : s'il devait etre coupe, qu'il le soit apres que tout le reste
|
|
a ete prouve — et qu'on le voie seul en cause.
|
|
"""
|
|
import yaml
|
|
enfants = (yaml.safe_load(_inventaire(instance).read_text(encoding="utf-8"))
|
|
.get("all", {}).get("children", {}))
|
|
runners = set()
|
|
for g in ("serveur_ops_tenant", "serveur_ops"):
|
|
runners |= set(((enfants.get(g) or {}).get("hosts") or {}))
|
|
hotes = sorted(_ip_par_hote(instance))
|
|
return [h for h in hotes if h not in runners] + [h for h in hotes if h in runners]
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--instance", required=True, help="le locataire, ex. OPS-Technolibre")
|
|
cible = ap.add_mutually_exclusive_group(required=True)
|
|
cible.add_argument("--hote", help="la VM, ex. idm-01")
|
|
cible.add_argument("--flotte", action="store_true",
|
|
help="toutes les VM, une a une, le runner en dernier ; arret au premier refus")
|
|
ap.add_argument("--sondes", action="store_true",
|
|
help="avec --flotte : tout activer d'un coup, juge par la sonde `connectivite` d'Icinga")
|
|
mode = ap.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--plan", action="store_true", help="verifier, sans rien activer")
|
|
mode.add_argument("--activer", action="store_true", help="verifier, activer, reverifier")
|
|
ap.add_argument("--runner", default="ansible@10.37.31.11", help="runner du site (API du cluster)")
|
|
ap.add_argument("--rebond", default="ansible@10.37.0.1", help="rebond vers le runner")
|
|
ap.add_argument("--racine-runner", default="/opt/setops")
|
|
args = ap.parse_args()
|
|
if args.hote:
|
|
return eprouver(args.instance, args.hote, args)
|
|
if args.sondes:
|
|
return activer_par_sondes(args.instance, args)
|
|
# UNE A UNE, ET ON S'ARRETE AU PREMIER REFUS (2026-09-30). C'etait une boucle tapee a
|
|
# la main apres chaque reconstruction ; la regle qui la rend sure — ne pas continuer
|
|
# apres un ecart — doit vivre avec elle, pas dans la memoire de celui qui la tape.
|
|
hotes = ordre_flotte(args.instance)
|
|
for n, h in enumerate(hotes, 1):
|
|
rc = eprouver(args.instance, h, args)
|
|
if rc != 0:
|
|
print(f"\nARRET sur {h} (code {rc}) : {n - 1}/{len(hotes)} VM traitees avant lui. "
|
|
"Rien d'autre n'a ete touche.")
|
|
return rc
|
|
print(f"\n{len(hotes)}/{len(hotes)} VM : pare-feu Proxmox "
|
|
+ ("verifiable" if args.plan else "actif") + ", aucun flux perdu.")
|
|
return 0
|
|
|
|
|
|
def eprouver(I: str, H: str, args: argparse.Namespace) -> int:
|
|
"""La procedure pour UNE VM (voir l'en-tete). 0 = conforme."""
|
|
tests, vmid, reseaux = matrice(I, H)
|
|
print(f"=== {H} ({I}, VM {vmid}) — {len(tests)} test(s) tires du devis")
|
|
hors = observer(I, H, tests, reseaux)
|
|
print(f" observation : {len(hors)} flux etabli(s) hors des regles")
|
|
for h in hors:
|
|
print(f" HORS : {h}")
|
|
avant = jouer(I, H, tests)
|
|
ecoute = ecoutes(I, H)
|
|
def _sans_objet(cle: str) -> bool:
|
|
proto, port = cle.rsplit(":", 1)[1].split("/", 1)
|
|
return proto != "icmp" and port not in ecoute
|
|
sans_objet = {k for k, v in avant.items() if v == "echec" and _sans_objet(k)}
|
|
echecs = sorted(k for k, v in avant.items() if v == "echec" and k not in sans_objet)
|
|
print(f" avant : {sum(v == 'ok' for v in avant.values())}/{len(avant)} ok"
|
|
+ (f", {len(sans_objet)} sans objet (rien n'ecoute hors de 127.0.0.1)" if sans_objet else ""))
|
|
for k in sorted(sans_objet):
|
|
print(f" sans objet : {k}")
|
|
for k in echecs:
|
|
print(f" DEJA EN ECHEC : {k}")
|
|
if args.plan:
|
|
print(" PLAN SEUL — rien n'a ete active.")
|
|
return 1 if (hors or echecs) else 0
|
|
if hors or echecs:
|
|
print(" REFUS D'ACTIVER : un flux reel hors des regles, ou un echec a comprendre d'abord.")
|
|
return 2
|
|
# LES CRITIQUES D'AVANT NE SONT PAS LES NOTRES (2026-09-30). Tout critique d'Icinga
|
|
# arretait la procedure — y compris ceux qui existaient avant l'activation : a la
|
|
# reconstruction de Chezlepro, des `restauration` rouges (rien a voir avec le pare-feu)
|
|
# l'ont arretee deux fois. On releve donc AVANT, et seul ce qui apparait apres compte.
|
|
crit_avant = set(critiques_icinga(I))
|
|
if activer(I, vmid, args) != 0:
|
|
print(" ECHEC de l'activation.")
|
|
return 3
|
|
print(f" active a {time.strftime('%H:%M:%S')}")
|
|
apres = jouer(I, H, tests)
|
|
change = sorted(k for k in avant if avant[k] != apres.get(k))
|
|
print(f" apres : {sum(v == 'ok' for v in apres.values())}/{len(apres)} ok ; "
|
|
f"changements : {', '.join(change) or 'aucun'}")
|
|
crit_apres = critiques_icinga(I)
|
|
crit = sorted(set(crit_apres) - crit_avant)
|
|
print(f" Icinga : {len(crit)} critique(s) apparu(s)" + (f" — {', '.join(crit)}" if crit else "")
|
|
+ (f" ; {len(crit_avant)} deja present(s) avant, sans rapport avec le pare-feu"
|
|
if crit_avant else ""))
|
|
if change:
|
|
print(f" RETOUR ARRIERE : remettre `enable=0` sur la VM {vmid} "
|
|
f"(/nodes/<noeud>/qemu/{vmid}/firewall/options).")
|
|
return 4 if (change or crit) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|