Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity
Set-OPS-Public/scripts/vpn_admin.py
Daniel Allaire 5f52dcb2b7 wireguard : rotation des cles de daniel-portable, sans afficher de privee 
cle-appareil et config --cle/--vers ecrivent les secrets en 0600 ;
--portee restreint appliquer au tunnel vise. Nouvelles cles en service,
anciennes retirees, Technolibre intact.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-28 12:57:15 -04:00

469 lines
22 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""L'acces d'administration par WireGuard — le plan declare, la frontiere suit.
POURQUOI CETTE VOIE (2026-09-17). L'exploitant entrait par les pattes de la frontiere, une
regle par zone. La question posee etait : « le runner ne devrait-il pas etre le rebond ? »
Non — il detient la cle de la voute du site et les cles SSH de toutes les machines : une
session humaine compromise deviendrait le plan de controle, et reconstruire le runner
couperait l'acces. Un tunnel nominatif donne la meme porte unique sans reunir ces pouvoirs.
CE QUE CE SCRIPT RECONCILIE, et rien d'autre :
- UNE instance WireGuard (celle que le plan nomme), distincte du tunnel site-a-site ;
- les PAIRS qui lui sont attaches : un par personne ET par appareil.
PERIMETRE STRICT : un pair attache a une AUTRE instance n'est jamais touche — le tunnel vers
le site pair vit a cote et ne doit rien craindre d'ici.
AUCUNE CLE PRIVEE D'APPAREIL N'ENTRE DANS LE DEPOT. Le plan ne porte que des cles PUBLIQUES.
`pair-nouveau` tire une paire, affiche la privee UNE fois (a coller dans l'appareil) et rend
la ligne de plan a ajouter.
Usage :
python3 scripts/vpn_admin.py plan # aucune ecriture
CONFIRMER=true python3 scripts/vpn_admin.py appliquer
python3 scripts/vpn_admin.py pair-nouveau --nom daniel-portable
python3 scripts/vpn_admin.py config --nom daniel-portable
"""
from __future__ import annotations
import argparse
import base64
import ipaddress
import os
import sys
from pathlib import Path
RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
import appliquer_opnsense as appl # noqa: E402
import devis_opnsense as devis_mod # noqa: E402
import underlay as underlay_mod # noqa: E402
def tunnels(inclure_vides: bool = False) -> list[dict]:
"""Tous les tunnels a tenir : celui du SITE, puis un par locataire qui declare des pairs.
DEUX DECLARANTS, DEUX PORTEES. Le site declare le sien dans son plan (`acces_admin_vpn`).
Un LOCATAIRE declare les siens dans SON plan (`plan/acces.yml`) : c'est lui qui decide qui
entre chez lui, et le site ne porte que la route — meme partage que les zones DNS
publiques. Son reseau, son port et son numero d'instance se DERIVENT de son index : rien
a allouer, aucune collision possible (P21 garde deja l'unicite des index).
"""
import devis_reseau
from inventory_rules import (charger_acces, charger_nomenclature, port_vpn_locataire,
reseau_vpn_locataire, valider_acces)
out: list[dict] = []
site = (underlay_mod.lire_plan_site("10-intrants.yml") or {}).get("acces_admin_vpn") or {}
if site.get("reseau") and site.get("port"):
out.append({**site,
"nom": str(site.get("nom") or "admins"),
"portee": "site",
"pairs": {str(p["nom"]): p for p in (site.get("pairs") or [])}})
for nom_depot, _prefixe, _n in devis_reseau.decouvrir_du_site():
base = RACINE.parent / nom_depot
registre = charger_acces(base / "plan" / "acces.yml")
index = charger_nomenclature(base / "plan" / "nomenclature.yml").get("index")
# UN LOCATAIRE SANS PAIR N'A PAS DE TUNNEL — sauf quand on vient justement lui en
# proposer un (`pair-nouveau`) : refuser la` aurait fait de la premiere cle un
# probleme d'oeuf et de poule, et c'est ce que le premier essai a montre.
if index is None or (not (registre.get("acces_admin_vpn") or {}) and not inclure_vides):
continue
valider_acces(registre, index)
slug = nom_depot.lower().removeprefix("ops-")
out.append({
"nom": f"admins-{slug}",
"portee": nom_depot,
# L'INSTANCE EST L'INDEX : `wg17`, `wg23`. Un rang (2, 3, 4...) changerait de
# numero le jour ou un locataire part, et renommerait l'interface des autres.
"instance": int(index),
"port": port_vpn_locataire(index),
"reseau": reseau_vpn_locataire(index),
"adresse_frontiere": f"{next(ipaddress.ip_network(reseau_vpn_locataire(index)).hosts())}/24",
"mtu": int(site.get("mtu", 1412)),
# Le nom du pair sur le boitier porte le locataire : deux ecosystemes peuvent
# avoir chacun un `daniel-portable` sans se marcher dessus.
"pairs": {f"{slug}-{n}": {"nom": f"{slug}-{n}", **c}
for n, c in (registre.get("acces_admin_vpn") or {}).items()},
})
return out
def plan_vpn() -> dict:
d = (underlay_mod.lire_plan_site("10-intrants.yml") or {}).get("acces_admin_vpn") or {}
if not d.get("reseau") or not d.get("port"):
raise SystemExit("REFUS : le plan du site ne declare pas `acces_admin_vpn` "
"(reseau, port). Rien a reconcilier.")
return d
def api_frontiere() -> appl.Frontiere:
base = devis_mod.depot_hebergeur()
if base is None:
raise SystemExit("Aucun underlay ne designe d'hebergeur : pas de frontiere a piloter.")
intr = devis_mod.intrants_frontiere()
v = appl._voute(base)
return appl.Frontiere(str(intr.get("opnsense_api_url") or ""),
v["vault_opnsense_api_key"], v["vault_opnsense_api_secret"],
bool(intr.get("opnsense_api_verifier_certs")))
def _table(rep: dict, *chemin: str) -> dict:
"""OPNsense emboite ses listes : {'server': {'servers': {'server': {uuid: {...}}}}}."""
noeud = rep
for c in chemin:
noeud = (noeud or {}).get(c) or {}
return noeud if isinstance(noeud, dict) else {}
def _choisi(champ) -> list[str]:
"""Un champ a choix d'OPNsense : rend les valeurs selectionnees."""
if isinstance(champ, dict):
return [k for k, v in champ.items() if isinstance(v, dict) and v.get("selected")]
return [str(champ)] if champ else []
def etat(api: appl.Frontiere) -> tuple[dict, dict]:
"""(serveurs par uuid, pairs par uuid) tels que la frontiere les porte."""
srv = _table(appl._lire(api, "/api/wireguard/server/get"), "server", "servers", "server")
pairs = _table(appl._lire(api, "/api/wireguard/client/get"), "client", "clients", "client")
return srv, pairs
def reseaux_joignables() -> list[str]:
"""Ce que le tunnel doit router : les zones du site, la fabric, les supernets locataires.
DERIVE, JAMAIS ECRIT. Une liste recopiee ici prendrait du retard sur la carte a la
premiere zone ajoutee — et l'administrateur decouvrirait le trou au pire moment.
"""
u = underlay_mod.charger() or {}
out = {str(r["sous_reseau"]) for r in underlay_mod.reseaux(u)
if r.get("sous_reseau")
and (str(r.get("nom", "")) in ("management", "grappe-controle", "transit-frontiere")
or str(r.get("nom", "")).startswith("site-"))}
out |= {str(s) for s in _supernets_locataires()}
return sorted(out, key=lambda c: ipaddress.ip_network(c, strict=False))
def _supernets_locataires() -> list[str]:
"""Les supernets des locataires heberges — decouverts, jamais listes."""
import devis_reseau
out = []
for nom_depot, _prefixe, _n in devis_reseau.decouvrir_du_site():
try:
nomencl = (RACINE.parent / nom_depot / "plan" / "nomenclature.yml")
import yaml
d = yaml.safe_load(nomencl.read_text(encoding="utf-8")) or {}
idx = d.get("index")
if idx is not None:
out.append(f"10.{int(idx)}.0.0/16")
except (OSError, ValueError, TypeError):
continue
return sorted(out)
def rapprocher(api: appl.Frontiere, vpn: dict) -> dict:
srv, pairs = etat(api)
nom = str(vpn.get("nom") or "admins")
uuid_srv = next((u for u, s in srv.items() if str(s.get("name")) == nom), None)
declares = {n: p for n, p in (vpn.get("pairs") or {}).items()
if str(p.get("etat", "present")) == "present"}
retires = {n for n, p in (vpn.get("pairs") or {}).items()
if str(p.get("etat", "present")) == "absent"}
poses = {}
if uuid_srv:
for u, p in pairs.items():
if uuid_srv in _choisi(p.get("servers")):
poses[str(p.get("name"))] = (u, p)
def different(nom_p: str) -> bool:
_u, p = poses[nom_p]
d = declares[nom_p]
return (str(p.get("pubkey")) != str(d.get("cle_publique"))
or _choisi(p.get("tunneladdress")) != [str(d.get("adresse"))])
return {
"nom": nom,
"uuid_serveur": uuid_srv,
"serveur_a_creer": uuid_srv is None,
"serveur": srv.get(uuid_srv or "", {}),
"pairs_creer": {n: d for n, d in declares.items() if n not in poses},
"pairs_majer": {n: d for n, d in declares.items() if n in poses and different(n)},
"pairs_garder": {n for n in declares if n in poses and not different(n)},
# UN PAIR ATTACHE A NOTRE INSTANCE QUE LE PLAN NE DECLARE PLUS : c'est un acces qui
# survit a la decision de le retirer. On le nomme, et `appliquer` le retire.
"pairs_retirer": {n: u for n, (u, _p) in poses.items()
if n not in declares or n in retires},
}
def afficher(p: dict, vpn: dict) -> bool:
if p["serveur_a_creer"]:
print(f" + instance WireGuard {p['nom']} (port {vpn['port']}, "
f"{vpn['adresse_frontiere']})")
else:
s = p["serveur"]
print(f" = instance WireGuard {p['nom']} — port {s.get('port')}, "
f"cle publique {str(s.get('pubkey'))[:12]}…")
for n, d in sorted(p["pairs_creer"].items()):
print(f" + pair {n:<24} {d.get('adresse')}")
for n, d in sorted(p["pairs_majer"].items()):
print(f" ~ pair MODIFIE {n:<24} {d.get('adresse')}")
for n in sorted(p["pairs_retirer"]):
print(f" - pair RETIRE {n}")
print(f"\n a creer : {len(p['pairs_creer']) + (1 if p['serveur_a_creer'] else 0)}"
f" | a retirer : {len(p['pairs_retirer'])}"
f" | inchange : {len(p['pairs_garder'])}")
return bool(p["serveur_a_creer"] or p["pairs_creer"] or p["pairs_majer"] or p["pairs_retirer"])
def _cle_privee() -> tuple[str, str]:
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from cryptography.hazmat.primitives import serialization
k = X25519PrivateKey.generate()
priv = k.private_bytes(encoding=serialization.Encoding.Raw,
format=serialization.PrivateFormat.Raw,
encryption_algorithm=serialization.NoEncryption())
pub = k.public_key().public_bytes(encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw)
return base64.b64encode(priv).decode(), base64.b64encode(pub).decode()
def appliquer(api: appl.Frontiere, p: dict, vpn: dict) -> int:
echecs = 0
def fait(rep, quoi):
nonlocal echecs
if rep.get("result") in ("saved", "deleted") or rep.get("status") == "ok":
return rep
echecs += 1
print(f" ! ECHEC {quoi} : {str(rep)[:200]}")
return {}
uuid_srv = p["uuid_serveur"]
if p["serveur_a_creer"]:
# LA CLE DU SERVEUR NAIT SUR LE BOITIER ET Y RESTE. Elle n'ouvre rien toute seule :
# c'est la cle de la FRONTIERE, pas celle d'un administrateur. Le jour ou le boitier
# est refait, les appareils recoivent la nouvelle cle publique — et c'est tout.
priv, pub = _cle_privee()
rep = fait(api("/api/wireguard/server/add_server", {"server": {
"enabled": "1", "name": p["nom"], "instance": str(vpn.get("instance", 1)),
"pubkey": pub, "privkey": priv, "port": str(vpn["port"]),
"mtu": str(vpn.get("mtu", 1412)),
"tunneladdress": str(vpn["adresse_frontiere"]),
"disableroutes": "0", "peers": "",
}}), f"instance {p['nom']}")
uuid_srv = rep.get("uuid") or None
if uuid_srv is None:
srv, _ = etat(api)
uuid_srv = next((u for u, s in srv.items() if str(s.get("name")) == p["nom"]), None)
for nom_p, d in sorted({**p["pairs_creer"], **p["pairs_majer"]}.items()):
corps = {"client": {
"enabled": "1", "name": nom_p, "pubkey": str(d["cle_publique"]), "psk": "",
"tunneladdress": str(d["adresse"]), "keepalive": "25",
"servers": uuid_srv or "",
}}
if nom_p in p["pairs_majer"]:
fait(api(f"/api/wireguard/client/set_client/{_uuid_pair(api, nom_p)}", corps),
f"pair {nom_p}")
else:
fait(api("/api/wireguard/client/add_client", corps), f"pair {nom_p}")
for nom_p, u in sorted(p["pairs_retirer"].items()):
fait(api(f"/api/wireguard/client/del_client/{u}", {}), f"retrait du pair {nom_p}")
if echecs:
print(f"\n {echecs} echec(s) — RIEN N'EST RECHARGE : le tunnel garde son etat "
f"precedent. Corriger, puis rejouer.")
return 1
# OPNsense repond `{"result": "ok"}` : lire `status` seul affichait « ? » apres un
# rechargement reussi (constate le 2026-09-28, nouvelles cles bien en service).
rep = api("/api/wireguard/service/reconfigure", {})
print(" service :", rep.get("status") or rep.get("result") or rep)
return 0
def _uuid_pair(api: appl.Frontiere, nom_p: str) -> str:
_srv, pairs = etat(api)
return next((u for u, x in pairs.items() if str(x.get("name")) == nom_p), "")
def cmd_plan(args) -> int:
api = api_frontiere()
tous = tunnels()
if not tous:
print("Aucun tunnel declare — ni au plan du site, ni chez un locataire.")
return 0
# N'APPLIQUER QUE CE QU'ON VISE (2026-09-28). Sans filtre, `appliquer` reconcilie TOUS
# les tunnels : une rotation de cle chez l'un aurait emporte l'ecart en attente d'un
# autre — un pair retire que personne n'avait encore decide de retirer.
if getattr(args, "portee", None):
voulues = {x.lower() for x in args.portee}
connues = {t["portee"].lower() for t in tous}
if voulues - connues:
raise SystemExit(f"REFUS : portee inconnue {sorted(voulues - connues)} "
f"(connues : {sorted(connues)}).")
tous = [t for t in tous if t["portee"].lower() in voulues]
plans, a_faire = [], False
for vpn in tous:
p = rapprocher(api, vpn)
plans.append((p, vpn))
print(f"Tunnel « {p['nom']} » ({vpn['portee']}) — reseau {vpn['reseau']}, "
f"port {vpn['port']}\n")
a_faire = afficher(p, vpn) or a_faire
print()
if not a_faire:
print(" Rien a faire.")
return 0
if os.environ.get("CONFIRMER") != "true":
print(" PLAN SEUL — aucune ecriture. Rejouer avec CONFIRMER=true pour appliquer.")
return 0
code = 0
for p, vpn in plans:
code = appliquer(api, p, vpn) or code
return code
def _tunnel_vise(portee: str | None) -> dict:
"""Le tunnel du site par defaut ; celui d'un locataire avec `--locataire OPS-X`."""
tous = tunnels(inclure_vides=True)
if portee:
for t in tous:
if t["portee"].lower() == portee.lower():
return t
raise SystemExit(f"REFUS : aucun tunnel declare pour « {portee} » "
f"(connus : {', '.join(t['portee'] for t in tous) or 'aucun'}).")
for t in tous:
if t["portee"] == "site":
return t
raise SystemExit("REFUS : le plan du site ne declare aucun tunnel.")
def cmd_pair(args) -> int:
vpn = _tunnel_vise(getattr(args, "locataire", None))
priv, pub = _cle_privee()
reseau = ipaddress.ip_network(str(vpn["reseau"]), strict=False)
prises = {str(p.get("adresse", "")).split("/")[0] for p in (vpn.get("pairs") or {}).values()}
prises.add(str(vpn["adresse_frontiere"]).split("/")[0])
libre = next((str(h) for h in reseau.hosts() if str(h) not in prises), None)
if vpn["portee"] == "site":
print(f"# Pair « {args.nom} » — a AJOUTER au plan du site (plan/10-intrants.yml) :\n")
print(" pairs:")
print(f" - nom: {args.nom}")
print(f" cle_publique: \"{pub}\"")
print(f" adresse: {libre}/32")
print(" etat: present\n")
else:
print(f"# Pair « {args.nom} » — a AJOUTER au plan de {vpn['portee']} (plan/acces.yml) :\n")
print("acces_admin_vpn:")
print(f" {args.nom}:")
print(f" cle_publique: \"{pub}\"")
print(f" adresse: {libre}/32")
print(" etat: present\n")
print("# La cle PRIVEE ci-dessous ne s'affiche qu'une fois : elle appartient a l'appareil,")
print("# elle n'entre ni dans le depot ni dans une voute.\n")
print(f"PrivateKey = {priv}\n")
print(f"# Puis : python3 scripts/vpn_admin.py config --nom {args.nom}")
return 0
def _ecrire_secret(chemin: Path, contenu: str) -> None:
"""Ecrire un fichier que seul son proprietaire lit, sans jamais en ecraser un."""
chemin = chemin.expanduser()
if chemin.exists():
raise SystemExit(f"REFUS : {chemin} existe deja. On n'ecrase pas une cle.")
chemin.parent.mkdir(parents=True, exist_ok=True)
fd = os.open(chemin, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
with os.fdopen(fd, "w") as fh:
fh.write(contenu)
def cmd_cle_appareil(args) -> int:
"""LA CLE PRIVEE NE PASSE PAR AUCUN ECRAN (2026-09-28).
`pair-nouveau` affiche la privee pour qu'un humain la colle. Lance par un assistant,
ou dans un terminal journalise, cet affichage la depose dans un historique — c'est
ainsi que deux cles d'administration ont du etre changees. Ici, la privee va droit
dans un fichier en 0600 ; seule la PUBLIQUE s'affiche, pour le plan. Sert aussi a une
ROTATION : on remplace `cle_publique` du pair existant, nom et adresse inchanges.
"""
priv, pub = _cle_privee()
_ecrire_secret(Path(args.vers), priv + "\n")
print(f"Cle privee ecrite dans {Path(args.vers).expanduser()} (0600, jamais affichee).")
print(f"cle_publique: \"{pub}\"")
return 0
def cmd_config(args) -> int:
vpn = _tunnel_vise(getattr(args, "locataire", None))
d = next((p for n, p in (vpn.get("pairs") or {}).items()
if n == args.nom or str(p.get("nom")) == args.nom
or n.endswith(f"-{args.nom}")), None)
if d is None:
raise SystemExit(f"REFUS : aucun pair « {args.nom} » au plan de {vpn['portee']}.")
api = api_frontiere()
srv, _pairs = etat(api)
nom = str(vpn["nom"])
s = next((x for x in srv.values() if str(x.get("name")) == nom), None)
if s is None:
raise SystemExit(f"REFUS : l'instance « {nom} » n'existe pas encore sur la frontiere "
f"(CONFIRMER=true python3 scripts/vpn_admin.py appliquer).")
intr = devis_mod.intrants_frontiere()
cle = (Path(args.cle).expanduser().read_text().strip() if getattr(args, "cle", None)
else "<cle privee de l'appareil>")
texte = (f"[Interface]\n# PrivateKey : celle affichee par `pair-nouveau` pour {args.nom}\n"
f"PrivateKey = {cle}\n"
f"Address = {d['adresse']}\n"
f"MTU = {vpn.get('mtu', 1412)}\n\n"
f"[Peer]\n"
f"PublicKey = {s.get('pubkey')}\n"
f"Endpoint = {intr.get('opnsense_wan_ip')}:{vpn['port']}\n"
f"AllowedIPs = {', '.join(reseaux_joignables() if vpn['portee'] == 'site' else [_supernet_de(vpn)])}\n"
f"PersistentKeepalive = 25\n")
if getattr(args, "vers", None):
if not getattr(args, "cle", None):
raise SystemExit("REFUS : --vers exige --cle (sinon le fichier porterait un gabarit).")
_ecrire_secret(Path(args.vers), texte)
print(f"Configuration complete ecrite dans {Path(args.vers).expanduser()} (0600).")
else:
print(texte, end="")
return 0
def _supernet_de(vpn: dict) -> str:
"""CE QU'UN TUNNEL DE LOCATAIRE ATTEINT : son ecosysteme, et rien d'autre."""
return f"10.{ipaddress.ip_network(str(vpn['reseau']), strict=False).network_address.packed[1]}.0.0/16"
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
sous = ap.add_subparsers(dest="commande", required=True)
for nom_cmd in ("plan", "appliquer"):
p = sous.add_parser(nom_cmd, help="rapproche le plan et la frontiere "
"(ecriture seulement avec CONFIRMER=true)")
p.add_argument("--portee", action="append",
help="restreindre a un tunnel : `site` ou `OPS-X` (repetable)")
p.set_defaults(fn=cmd_plan)
p = sous.add_parser("pair-nouveau", help="tire une paire de cles et rend la ligne de plan")
p.add_argument("--nom", required=True, help="personne-appareil, ex. daniel-portable")
p.add_argument("--locataire", help="le tunnel de cet ecosysteme (ex. OPS-Chezlepro) ; "
"sans lui, celui du site")
p.set_defaults(fn=cmd_pair)
p = sous.add_parser("cle-appareil", help="tire une paire : privee dans un FICHIER, "
"publique a l'ecran (creation ou rotation)")
p.add_argument("--vers", required=True, help="fichier de la cle privee (cree en 0600)")
p.set_defaults(fn=cmd_cle_appareil)
p = sous.add_parser("config", help="la configuration a coller dans l'appareil")
p.add_argument("--nom", required=True)
p.add_argument("--locataire")
p.add_argument("--cle", help="fichier de la cle privee (sinon un gabarit a completer)")
p.add_argument("--vers", help="ecrire la configuration complete dans ce fichier (0600)")
p.set_defaults(fn=cmd_config)
args = ap.parse_args()
return args.fn(args)
if __name__ == "__main__":
sys.exit(main())
Reference in a new issue View git blame Copy permalink