Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Set-OPS-Public/scripts/appliquer_proxmox_fw.py
Daniel Allaire d61c187dcf proxmox-fw : le DROP se pose par VM, jamais au datacenter (D-64) 
Le devis enseignait « politique d'entree DROP au datacenter ». Or policy_in y
est la politique par defaut de TOUTE VM dont le pare-feu s'active — 37 machines
heritees sans regles, sur ce cluster. `policy_in` existe aussi par VM : le devis
et l'applicateur le posent la. Meme isolation, sans falaise, et l'applicateur
n'a plus a refuser une partie de son devis.

Trois verrous, pas un : datacenter enable=1, `enable` de la VM (defaut 0), et
`firewall=1` sur la carte. C'est le verrou du milieu que j'avais manque en
annoncant que huit VM en production tomberaient.

`enable=1` au datacenter bascule et verifie : pve-firewall running, 12 chaines
cadres, AUCUNE chaine par VM, 0 regle visant roxanne, 15 VM toujours en marche,
hyperviseurs et frontiere joignables, sortie tenant 2/2.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 20:03:41 -04:00

269 lines
13 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Reconcilie le pare-feu Proxmox (est-ouest intra-tenant) avec son devis.
Trois couches, toutes au niveau du CLUSTER sauf la derniere :
- IPSets : qui porte quel role
- Groupes : les flux entrants d'un role
- Affectations : quels groupes sur quelle VM
DEUX SENS : ce que le devis demande et qui manque est cree, ce qu'il ne demande
plus est RETIRE. Un groupe orphelin ne bloque rien, mais il decrit une politique
qui n'est plus la notre.
LE DROP EST PORTE PAR LA VM, jamais par le datacenter. Au datacenter, `policy_in`
est la politique par defaut de TOUTE VM dont le pare-feu s'active — un parc herite
sans regles tomberait d'un bloc. Pose sur chaque VM tenant, il donne la meme
isolation sans cette falaise, et le script le fait lui-meme.
Il n'active pas le pare-feu du datacenter (`enable=1`) : c'est un reglage de
cluster, decide une fois, et signale tant qu'il manque.
PERIMETRE STRICT : seuls les objets prefixes par l'etiquette d'un tenant du devis
(`t11-`, `t17-`) sont touches. Le reste du cluster n'existe pas pour ce script.
NON DESTRUCTIF PAR DEFAUT : sans `CONFIRMER=true`, aucune ecriture (regle 4).
Usage :
python3 scripts/appliquer_proxmox_fw.py # plan seul
CONFIRMER=true python3 scripts/appliquer_proxmox_fw.py # applique
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
from proxmox_api import Cluster # noqa: E402
def _prefixe(bloc: dict) -> str:
"""`t11-` — deduit du premier IPSet, qui les porte tous. Jamais saisi."""
for nom in bloc.get("ipsets") or {}:
return nom.split("-")[0] + "-"
for g in bloc.get("groupes") or []:
return str(g["nom"]).split("-")[0] + "-"
return ""
def _regle_cle(r: dict) -> tuple:
"""Identite d'une regle de groupe, insensible a la position et au commentaire."""
return (str(r.get("type") or r.get("sens") or "").lower(),
str(r.get("action") or "").upper(),
str(r.get("source") or ""),
str(r.get("proto") or ""),
str(r.get("dport") or ""))
def plan(api: Cluster, devis: dict) -> dict:
prefixes = tuple(_prefixe(b) for b in devis["blocs"] if _prefixe(b))
ipsets_voulus, groupes_voulus, affect_voulues = {}, {}, {}
for b in devis["blocs"]:
for nom, d in (b.get("ipsets") or {}).items():
ipsets_voulus[nom] = {"membres": sorted(set(d["membres"])), "role": d.get("role", "")}
for g in b.get("groupes") or []:
groupes_voulus[g["nom"]] = g
for a in b.get("affectations") or []:
affect_voulues[int(a["vmid"])] = {"hote": a["hote"], "groupes": sorted(a["groupes"])}
# --- etat du cluster, restreint a nos prefixes -----------------------------
ipsets_poses = {}
for s in (api("/cluster/firewall/ipset") or []):
if isinstance(s, dict) and str(s.get("name", "")).startswith(prefixes):
membres = [str(m.get("cidr")) for m in (api(f"/cluster/firewall/ipset/{s['name']}") or [])
if isinstance(m, dict)]
ipsets_poses[s["name"]] = sorted(set(membres))
groupes_poses = {}
for g in (api("/cluster/firewall/groups") or []):
if isinstance(g, dict) and str(g.get("group", "")).startswith(prefixes):
regles = [r for r in (api(f"/cluster/firewall/groups/{g['group']}") or [])
if isinstance(r, dict)]
groupes_poses[g["group"]] = regles
vm_du_cluster = {int(r["vmid"]): r["node"] for r in (api("/cluster/resources?type=vm") or [])
if isinstance(r, dict) and r.get("vmid")}
affect_posees, vm_absentes = {}, []
for vmid in affect_voulues:
noeud = vm_du_cluster.get(vmid)
if not noeud:
vm_absentes.append(vmid)
continue
regles = api(f"/nodes/{noeud}/qemu/{vmid}/firewall/rules") or []
o = api(f"/nodes/{noeud}/qemu/{vmid}/firewall/options") or {}
affect_posees[vmid] = {
"noeud": noeud,
"groupes": sorted(str(r.get("action")) for r in regles
if isinstance(r, dict) and r.get("type") == "group"),
"actif": str(o.get("enable", "")),
# Le DROP vit ICI, sur la VM du tenant — pas au datacenter, ou il vaudrait
# pour tout le parc, y compris ce qui n'appartient a aucun tenant.
"politique": str(o.get("policy_in", "")).upper(),
}
opts = api("/cluster/firewall/options") or {}
return {
"ipsets_creer": {n: v for n, v in ipsets_voulus.items() if n not in ipsets_poses},
"ipsets_majer": {n: (ipsets_poses[n], v["membres"]) for n, v in ipsets_voulus.items()
if n in ipsets_poses and ipsets_poses[n] != v["membres"]},
"ipsets_retirer": sorted(n for n in ipsets_poses if n not in ipsets_voulus),
"groupes_creer": {n: g for n, g in groupes_voulus.items() if n not in groupes_poses},
"groupes_majer": {n: g for n, g in groupes_voulus.items()
if n in groupes_poses
and {_regle_cle(r) for r in g["regles"]}
!= {_regle_cle(r) for r in groupes_poses[n]}},
"groupes_retirer": sorted(n for n in groupes_poses if n not in groupes_voulus),
"affect_majer": {v: a for v, a in affect_voulues.items()
if v in affect_posees
and (affect_posees[v]["groupes"] != a["groupes"]
or affect_posees[v]["actif"] != "1"
or affect_posees[v]["politique"] != "DROP")},
"affect_posees": affect_posees,
"vm_absentes": sorted(vm_absentes),
"datacenter": {"actif": str(opts.get("enable", "")), "politique": str(opts.get("policy_in", ""))},
"_voulus": {"ipsets": ipsets_voulus, "groupes": groupes_voulus, "affect": affect_voulues},
}
def afficher(p: dict) -> bool:
for n in sorted(p["ipsets_creer"]):
print(f" + IPSet {n:<22} {len(p['ipsets_creer'][n]['membres'])} membre(s)")
for n, (avant, apres) in sorted(p["ipsets_majer"].items()):
print(f" ~ IPSet {n:<22} {len(avant)} -> {len(apres)} membre(s)")
for n in p["ipsets_retirer"]:
print(f" - IPSet PERIME {n}")
for n, g in sorted(p["groupes_creer"].items()):
print(f" + groupe {n:<22} {len(g['regles'])} regle(s) # {g.get('role','')}")
for n in sorted(p["groupes_majer"]):
print(f" ~ groupe {n:<22} regles differentes")
for n in p["groupes_retirer"]:
print(f" - groupe PERIME {n}")
for v, a in sorted(p["affect_majer"].items()):
print(f" ~ VM {v:<12} {a['hote']:<16} -> {', '.join(a['groupes'])}")
if p["vm_absentes"]:
print(f"\n {len(p['vm_absentes'])} VM du devis absente(s) du cluster — affectation "
f"differee, elle se posera au clonage :")
print(f" {', '.join(str(v) for v in p['vm_absentes'][:8])}"
+ (" …" if len(p["vm_absentes"]) > 8 else ""))
dc = p["datacenter"]
if dc["actif"] != "1":
print(f"\n /!\\ PREREQUIS NON REMPLI — pare-feu du datacenter desactive "
f"(enable={dc['actif'] or '0'}).")
print(" Sans lui, RIEN ne filtre : les objets ci-dessus restent inertes.")
print(" `enable=1` seul ne ferme rien — une VM n'est filtree que si SON propre")
print(" `enable` vaut 1 (defaut 0) et que sa carte porte `firewall=1`.")
print(" Ce script ne le bascule pas : c'est un reglage de cluster, decide une fois.")
if dc["politique"].upper() in ("DROP", "REJECT"):
print(f"\n /!\\ DANGER — `policy_in={dc['politique']}` AU DATACENTER.")
print(" C'est la politique par defaut de TOUTE VM dont le pare-feu s'active,")
print(" y compris ce qui n'appartient a aucun tenant. Set-OPS pose le DROP")
print(" PAR VM tenant : cette valeur au datacenter n'est pas requise, et expose")
print(" le reste du parc. La remettre a ACCEPT.")
creer = len(p["ipsets_creer"]) + len(p["groupes_creer"])
majer = len(p["ipsets_majer"]) + len(p["groupes_majer"]) + len(p["affect_majer"])
retirer = len(p["ipsets_retirer"]) + len(p["groupes_retirer"])
print(f"\n a creer : {creer} | a mettre a jour : {majer} | a retirer : {retirer}")
return bool(creer or majer or retirer)
def appliquer(api: Cluster, p: dict) -> int:
echecs = 0
def _fait(rep, quoi):
nonlocal echecs
if (e := api.rate(rep)):
echecs += 1
print(f" ! ECHEC {quoi} : {e[:150]}")
return False
return True
# 1. IPSets AVANT les groupes : une regle qui reference un IPSet absent est refusee.
for n, v in sorted(p["ipsets_creer"].items()):
if _fait(api("/cluster/firewall/ipset", "POST",
{"name": n, "comment": v["role"][:100]}), f"ipset {n}"):
for m in v["membres"]:
_fait(api(f"/cluster/firewall/ipset/{n}", "POST", {"cidr": m}), f"{n} += {m}")
for n, (avant, apres) in sorted(p["ipsets_majer"].items()):
for m in sorted(set(apres) - set(avant)):
_fait(api(f"/cluster/firewall/ipset/{n}", "POST", {"cidr": m}), f"{n} += {m}")
for m in sorted(set(avant) - set(apres)):
_fait(api(f"/cluster/firewall/ipset/{n}/{m}", "DELETE"), f"{n} -= {m}")
# 2. Groupes. Une mise a jour REMPLACE les regles : elles sont a nous, et les
# comparer une a une pour un edit partiel couterait plus que de les reposer.
for n in sorted(p["groupes_majer"]):
for r in reversed(api(f"/cluster/firewall/groups/{n}") or []):
_fait(api(f"/cluster/firewall/groups/{n}/{r['pos']}", "DELETE"), f"purge {n}")
for n, g in sorted({**p["groupes_creer"], **p["groupes_majer"]}.items()):
if n in p["groupes_creer"]:
_fait(api("/cluster/firewall/groups", "POST",
{"group": n, "comment": str(g.get("role", ""))[:100]}), f"groupe {n}")
for r in g["regles"]:
corps = {"type": str(r.get("sens", "in")).lower(), "action": r["action"],
"enable": 1, "comment": str(r.get("via") or g.get("role", ""))[:100]}
for cle, champ in (("source", "source"), ("proto", "proto"), ("dport", "dport")):
if r.get(cle):
corps[champ] = r[cle]
_fait(api(f"/cluster/firewall/groups/{n}", "POST", corps), f"regle de {n}")
# 3. Affectations aux VM, et activation de leur pare-feu — sans quoi les groupes
# seraient poses sur une machine qui ne filtre pas.
for v, a in sorted(p["affect_majer"].items()):
noeud = p["affect_posees"][v]["noeud"]
for r in reversed(api(f"/nodes/{noeud}/qemu/{v}/firewall/rules") or []):
if isinstance(r, dict) and r.get("type") == "group":
_fait(api(f"/nodes/{noeud}/qemu/{v}/firewall/rules/{r['pos']}", "DELETE"),
f"purge VM {v}")
for g in a["groupes"]:
_fait(api(f"/nodes/{noeud}/qemu/{v}/firewall/rules", "POST",
{"type": "group", "action": g, "enable": 1}), f"VM {v} += {g}")
# `policy_in=DROP` sur LA VM : c'est ce qui rend les groupes ci-dessus exclusifs.
# Au datacenter, la meme valeur s'appliquerait a tout ce qui n'est pas un tenant.
_fait(api(f"/nodes/{noeud}/qemu/{v}/firewall/options", "PUT",
{"enable": 1, "policy_in": "DROP"}), f"pare-feu de la VM {v}")
# 4. Retraits : groupes avant IPSets, un IPSet reference ne se supprime pas.
for n in p["groupes_retirer"]:
for r in reversed(api(f"/cluster/firewall/groups/{n}") or []):
_fait(api(f"/cluster/firewall/groups/{n}/{r['pos']}", "DELETE"), f"purge {n}")
_fait(api(f"/cluster/firewall/groups/{n}", "DELETE"), f"retrait groupe {n}")
for n in p["ipsets_retirer"]:
_fait(api(f"/cluster/firewall/ipset/{n}", "DELETE", {"force": 1}), f"retrait ipset {n}")
if echecs:
print(f"\n {echecs} echec(s) — la reconciliation est INCOMPLETE.")
return 1
print("\n Pare-feu est-ouest reconcilie.")
return 0
def main() -> int:
api, base = Cluster.depuis_hebergeur()
r = subprocess.run([sys.executable, str(RACINE / "scripts" / "devis_proxmox_fw.py"), "--json"],
cwd=RACINE, capture_output=True, text=True,
env={**os.environ, "SETOPS_INSTANCE": str(base)})
if r.returncode != 0:
raise SystemExit("Le devis ne se genere pas :\n" + r.stderr.strip()[:400])
devis = json.loads(r.stdout)
print(f"Cluster {api.hote} — {len(devis['blocs'])} tenant(s) au devis\n")
p = plan(api, devis)
quelque_chose = afficher(p)
if not quelque_chose:
print("\n Le cluster dit deja ce que le devis dit. Rien a faire.")
return 0
if os.environ.get("CONFIRMER") != "true":
print("\n PLAN SEUL — aucune ecriture. Rejouer avec CONFIRMER=true pour appliquer.")
return 0
print()
return appliquer(api, p)
if __name__ == "__main__":
sys.exit(main())
Reference in a new issue View git blame Copy permalink