322 lines
14 KiB
Python
322 lines
14 KiB
Python
|
#!/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 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 = {str(p["nom"]): p for p in (vpn.get("pairs") or [])
|
||
|
|
if str(p.get("etat", "present")) == "present"}
|
||
|
|
retires = {str(p["nom"]) for p in (vpn.get("pairs") or [])
|
||
|
|
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
|
||
|
|
print(" service :", api("/api/wireguard/service/reconfigure", {}).get("status", "?"))
|
||
|
|
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:
|
||
|
|
vpn = plan_vpn()
|
||
|
|
api = api_frontiere()
|
||
|
|
p = rapprocher(api, vpn)
|
||
|
|
print(f"Acces d'administration WireGuard — instance « {p['nom'] } », "
|
||
|
|
f"reseau {vpn['reseau']}, port {vpn['port']}\n")
|
||
|
|
a_faire = afficher(p, vpn)
|
||
|
|
if not a_faire:
|
||
|
|
print("\n 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
|
||
|
|
return appliquer(api, p, vpn)
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_pair(args) -> int:
|
||
|
|
vpn = plan_vpn()
|
||
|
|
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 [])}
|
||
|
|
prises.add(str(vpn["adresse_frontiere"]).split("/")[0])
|
||
|
|
libre = next((str(h) for h in reseau.hosts() if str(h) not in prises), None)
|
||
|
|
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")
|
||
|
|
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 cmd_config(args) -> int:
|
||
|
|
vpn = plan_vpn()
|
||
|
|
d = next((p for p in (vpn.get("pairs") or []) if str(p.get("nom")) == args.nom), None)
|
||
|
|
if d is None:
|
||
|
|
raise SystemExit(f"REFUS : aucun pair « {args.nom} » au plan du site.")
|
||
|
|
api = api_frontiere()
|
||
|
|
srv, _pairs = etat(api)
|
||
|
|
nom = str(vpn.get("nom") or "admins")
|
||
|
|
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()
|
||
|
|
print(f"[Interface]\n# PrivateKey : celle affichee par `pair-nouveau` pour {args.nom}\n"
|
||
|
|
f"PrivateKey = <cle privee de l'appareil>\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())}\n"
|
||
|
|
f"PersistentKeepalive = 25")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
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.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.set_defaults(fn=cmd_pair)
|
||
|
|
p = sous.add_parser("config", help="la configuration a coller dans l'appareil")
|
||
|
|
p.add_argument("--nom", required=True)
|
||
|
|
p.set_defaults(fn=cmd_config)
|
||
|
|
args = ap.parse_args()
|
||
|
|
return args.fn(args)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|