321 lines
14 KiB
Python
321 lines
14 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Les cles DNSSEC d'un locataire : les creer dans SA voute, en calculer les DS, les verifier.
|
||
|
|
|
||
|
|
POURQUOI LA CLE VIT DANS LA VOUTE, ET PAS SUR LA MACHINE (2026-09-16). PowerDNS range ses cles
|
||
|
|
dans une base SQLite sur le primaire. Or ce depot reconstruit ses machines depuis zero, et
|
||
|
|
c'est prouve. Une cle nee sur la machine mourrait avec elle : la nouvelle zone serait signee
|
||
|
|
par une AUTRE cle, le DS publie chez le registraire ne correspondrait plus, et tout resolveur
|
||
|
|
validant — la majorite de l'Internet — declarerait le domaine BOGUS. Le site web ne
|
||
|
|
repondrait plus, le courriel ne partirait plus, et rien sur nos machines ne serait en panne.
|
||
|
|
|
||
|
|
La cle est donc tiree ici, une fois, rangee dans la voute DU LOCATAIRE (c'est lui qui signe ;
|
||
|
|
l'hebergeur ne la detient pas), et le role `serveur_powerdns` l'importe a chaque passage.
|
||
|
|
|
||
|
|
LE DS SE CALCULE SANS LA MACHINE. Il se deduit de la cle publique ; ce script le rend depuis
|
||
|
|
la voute seule. Le locataire peut donc le remettre a son registraire avant meme que la zone
|
||
|
|
soit deployee — et le rapprocher de ce que le registre publie (`verifier`).
|
||
|
|
|
||
|
|
UNE CLE, CSK, ECDSA P-256 (algorithme 13). Un seul type de cle : pas de roulement ZSK a
|
||
|
|
orchestrer. P-256 : ce que tous les resolveurs valident, et des reponses courtes.
|
||
|
|
|
||
|
|
Usage :
|
||
|
|
python3 scripts/dnssec.py ds # DS de chaque zone `dnssec: true`
|
||
|
|
python3 scripts/dnssec.py verifier # voute, plan et registre concordent-ils ?
|
||
|
|
python3 scripts/dnssec.py generer chezlepro.ca # cree la cle si elle manque
|
||
|
|
CONFIRMER=true python3 scripts/dnssec.py generer chezlepro.ca --remplacer
|
||
|
|
|
||
|
|
Remplacer une cle dont le DS est publie rend le domaine BOGUS : `--remplacer` le refuse tant
|
||
|
|
que le registre publie un DS pour la zone.
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import base64
|
||
|
|
import datetime as dt
|
||
|
|
import hashlib
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import shutil
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
|
from inventory_rules import instance_courante # noqa: E402
|
||
|
|
from voutes import cle_de # noqa: E402
|
||
|
|
|
||
|
|
ALGORITHME = 13 # ECDSAP256SHA256
|
||
|
|
DRAPEAUX_CSK = 257 # SEP : la cle que le parent designe
|
||
|
|
RESOLVEUR = "9.9.9.9"
|
||
|
|
|
||
|
|
|
||
|
|
def nom_variable(zone: str) -> str:
|
||
|
|
"""`chezlepro.ca` -> `vault_dnssec_chezlepro_ca`. Le role derive le meme nom."""
|
||
|
|
return "vault_dnssec_" + re.sub(r"[^a-z0-9]", "_", zone.lower().rstrip("."))
|
||
|
|
|
||
|
|
|
||
|
|
# --- LA CLE ET SES DERIVES -------------------------------------------------------------
|
||
|
|
|
||
|
|
def _privee(scalaire_b64: str):
|
||
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
||
|
|
d = int.from_bytes(base64.b64decode(scalaire_b64), "big")
|
||
|
|
return ec.derive_private_key(d, ec.SECP256R1())
|
||
|
|
|
||
|
|
|
||
|
|
def nouvelle_cle() -> str:
|
||
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
||
|
|
k = ec.generate_private_key(ec.SECP256R1())
|
||
|
|
return base64.b64encode(k.private_numbers().private_value.to_bytes(32, "big")).decode()
|
||
|
|
|
||
|
|
|
||
|
|
def dnskey_rdata(scalaire_b64: str) -> bytes:
|
||
|
|
pub = _privee(scalaire_b64).public_key().public_numbers()
|
||
|
|
cle = pub.x.to_bytes(32, "big") + pub.y.to_bytes(32, "big")
|
||
|
|
return DRAPEAUX_CSK.to_bytes(2, "big") + bytes([3, ALGORITHME]) + cle
|
||
|
|
|
||
|
|
|
||
|
|
def etiquette(rdata: bytes) -> int:
|
||
|
|
"""Key tag, RFC 4034 annexe B."""
|
||
|
|
acc = 0
|
||
|
|
for i, octet in enumerate(rdata):
|
||
|
|
acc += octet << 8 if i % 2 == 0 else octet
|
||
|
|
acc = (acc & 0xFFFF) + (acc >> 16)
|
||
|
|
return acc & 0xFFFF
|
||
|
|
|
||
|
|
|
||
|
|
def _nom_filaire(zone: str) -> bytes:
|
||
|
|
return b"".join(bytes([len(p)]) + p.encode() for p in zone.lower().rstrip(".").split(".")) + b"\0"
|
||
|
|
|
||
|
|
|
||
|
|
def ds(zone: str, scalaire_b64: str) -> str:
|
||
|
|
"""Le DS (SHA-256, type 2) a remettre au registraire."""
|
||
|
|
rdata = dnskey_rdata(scalaire_b64)
|
||
|
|
empreinte = hashlib.sha256(_nom_filaire(zone) + rdata).hexdigest()
|
||
|
|
return f"{etiquette(rdata)} {ALGORITHME} 2 {empreinte}"
|
||
|
|
|
||
|
|
|
||
|
|
def fichier_prive(scalaire_b64: str) -> str:
|
||
|
|
"""Le format que `pdnsutil import-zone-key` lit (celui de BIND, v1.2)."""
|
||
|
|
return ("Private-key-format: v1.2\nAlgorithm: 13 (ECDSAP256SHA256)\n"
|
||
|
|
f"PrivateKey: {scalaire_b64}\n")
|
||
|
|
|
||
|
|
|
||
|
|
# --- LA VOUTE DU LOCATAIRE --------------------------------------------------------------
|
||
|
|
|
||
|
|
ENV = dict(os.environ, PAGER="cat", ANSIBLE_PAGER="cat")
|
||
|
|
for _k in ("ANSIBLE_VAULT_IDENTITY_LIST", "ANSIBLE_VAULT_PASSWORD_FILE"):
|
||
|
|
ENV.pop(_k, None) # une voute, UNE cle : on ne laisse rien d'autre essayer
|
||
|
|
|
||
|
|
|
||
|
|
def _instance() -> Path:
|
||
|
|
base = instance_courante()
|
||
|
|
if not base.exists():
|
||
|
|
raise SystemExit("REFUS : aucune instance montee (symlink `instance/` ou SETOPS_INSTANCE). "
|
||
|
|
"Une cle DNSSEC appartient a UN locataire.")
|
||
|
|
return base.resolve()
|
||
|
|
|
||
|
|
|
||
|
|
def _voute(base: Path) -> tuple[Path, Path]:
|
||
|
|
f = base / "inventories" / "principal" / "group_vars" / "all" / "vault.yml"
|
||
|
|
cle = cle_de(base.name)
|
||
|
|
if not f.is_file() or not cle.is_file():
|
||
|
|
raise SystemExit(f"REFUS : voute ({f}) ou cle ({cle}) absente.")
|
||
|
|
return f, cle
|
||
|
|
|
||
|
|
|
||
|
|
def _dechiffrer(fichier: Path, cle: Path) -> str:
|
||
|
|
# PAR UN TUBE : ansible-vault ecrit sur une sortie non bloquante et echoue en silence
|
||
|
|
# sans lui (docs : « ansible-vault exige un tube »).
|
||
|
|
r = subprocess.run(["ansible-vault", "view", "--vault-password-file", str(cle), str(fichier)],
|
||
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=ENV)
|
||
|
|
if r.returncode != 0:
|
||
|
|
raise SystemExit(f"ECHEC du dechiffrement de {fichier} : {r.stderr.strip()[-160:]}")
|
||
|
|
return r.stdout
|
||
|
|
|
||
|
|
|
||
|
|
def lire_voute(base: Path) -> dict:
|
||
|
|
return yaml.safe_load(_dechiffrer(*_voute(base))) or {}
|
||
|
|
|
||
|
|
|
||
|
|
def _empreinte(d: dict) -> str:
|
||
|
|
return hashlib.sha256(yaml.safe_dump(d, sort_keys=True, allow_unicode=True).encode()).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def ecrire_variable(base: Path, nom: str, valeur: str, commentaire: str) -> Path:
|
||
|
|
"""Pose ou remplace UNE variable, voute relue avant et apres. Rend la copie de surete."""
|
||
|
|
f, cle = _voute(base)
|
||
|
|
clair = _dechiffrer(f, cle)
|
||
|
|
data = yaml.safe_load(clair) or {}
|
||
|
|
motif = re.compile(rf"^{re.escape(nom)}:.*$", re.MULTILINE)
|
||
|
|
if nom in data:
|
||
|
|
if len(motif.findall(clair)) != 1:
|
||
|
|
raise SystemExit(f"REFUS : `{nom}` n'est pas sur une seule ligne dans la voute ; "
|
||
|
|
"a remplacer a la main.")
|
||
|
|
nouveau = motif.sub(f'{nom}: "{valeur}"', clair)
|
||
|
|
else:
|
||
|
|
nouveau = clair.rstrip("\n") + f"\n\n# --- {commentaire} ---\n{nom}: \"{valeur}\"\n"
|
||
|
|
attendu = {**data, nom: valeur}
|
||
|
|
if yaml.safe_load(nouveau) != attendu:
|
||
|
|
raise SystemExit("REFUS : le YAML modifie ne reproduit pas l'attendu ; rien n'est ecrit.")
|
||
|
|
|
||
|
|
sauve = Path.home() / ".config" / f"setops-voutes-avant-dnssec-{dt.date.today().isoformat()}"
|
||
|
|
sauve.mkdir(mode=0o700, exist_ok=True)
|
||
|
|
copie = sauve / f"{base.name}--{dt.datetime.now():%H%M%S}--{f.name}"
|
||
|
|
shutil.copy2(f, copie)
|
||
|
|
|
||
|
|
atelier = Path(tempfile.mkdtemp(dir=sauve))
|
||
|
|
atelier.chmod(0o700)
|
||
|
|
f_clair, f_chiffre = atelier / "clair.yml", atelier / "chiffre.yml"
|
||
|
|
try:
|
||
|
|
f_clair.write_text(nouveau, encoding="utf-8")
|
||
|
|
f_clair.chmod(0o600)
|
||
|
|
r = subprocess.run(["ansible-vault", "encrypt", "--vault-password-file", str(cle),
|
||
|
|
"--output", str(f_chiffre), str(f_clair)],
|
||
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=ENV)
|
||
|
|
if r.returncode != 0:
|
||
|
|
raise SystemExit(f"ECHEC du chiffrement : {r.stderr.strip()[-160:]}")
|
||
|
|
if not f_chiffre.read_text(encoding="utf-8").startswith("$ANSIBLE_VAULT;1.1;AES256"):
|
||
|
|
raise SystemExit("ECHEC : en-tete de voute inattendu ; rien n'est remplace.")
|
||
|
|
if _empreinte(yaml.safe_load(_dechiffrer(f_chiffre, cle)) or {}) != _empreinte(attendu):
|
||
|
|
raise SystemExit("ECHEC : la relecture differe de l'attendu ; rien n'est remplace.")
|
||
|
|
mode = f.stat().st_mode & 0o777
|
||
|
|
os.replace(f_chiffre, f)
|
||
|
|
f.chmod(mode)
|
||
|
|
finally:
|
||
|
|
if f_clair.exists():
|
||
|
|
subprocess.run(["shred", "-u", "-z", str(f_clair)], check=False)
|
||
|
|
shutil.rmtree(atelier, ignore_errors=True)
|
||
|
|
if _empreinte(yaml.safe_load(_dechiffrer(f, cle)) or {}) != _empreinte(attendu):
|
||
|
|
raise SystemExit(f"ECHEC : la voute en place differe apres remplacement. Copie : {copie}")
|
||
|
|
return copie
|
||
|
|
|
||
|
|
|
||
|
|
# --- LE PLAN ET LE REGISTRE -------------------------------------------------------------
|
||
|
|
|
||
|
|
def zones_signees(base: Path) -> list[str]:
|
||
|
|
reg = yaml.safe_load((base / "plan" / "domaines.yml").read_text(encoding="utf-8")) or {}
|
||
|
|
return sorted(z for z, c in (reg.get("domaines_publics") or {}).items()
|
||
|
|
if c.get("autorite") == "primaire-cache" and c.get("dnssec") is True)
|
||
|
|
|
||
|
|
|
||
|
|
def ds_publies(zone: str) -> set[str] | None:
|
||
|
|
"""Les DS que le parent publie, ou None si la mesure echoue (jamais « aucun »)."""
|
||
|
|
r = subprocess.run(["dig", "+noall", "+answer", "+comments", "+time=4", "+tries=2",
|
||
|
|
f"@{RESOLVEUR}", zone, "DS"], capture_output=True, text=True, timeout=30)
|
||
|
|
statut = next((ligne.split("status:")[1].split(",")[0].strip()
|
||
|
|
for ligne in r.stdout.splitlines() if "status:" in ligne), None)
|
||
|
|
if r.returncode != 0 or statut != "NOERROR":
|
||
|
|
return None
|
||
|
|
vus = set()
|
||
|
|
for ligne in r.stdout.splitlines():
|
||
|
|
champs = ligne.split()
|
||
|
|
if len(champs) >= 8 and not ligne.startswith(";") and champs[3] == "DS":
|
||
|
|
vus.add(f"{champs[4]} {champs[5]} {champs[6]} {''.join(champs[7:]).lower()}")
|
||
|
|
return vus
|
||
|
|
|
||
|
|
|
||
|
|
# --- COMMANDES ------------------------------------------------------------------------
|
||
|
|
|
||
|
|
def cmd_ds(args) -> int:
|
||
|
|
base = _instance()
|
||
|
|
voute = lire_voute(base)
|
||
|
|
zones = [args.zone] if args.zone else zones_signees(base)
|
||
|
|
if not zones:
|
||
|
|
print(f"Aucune zone `dnssec: true` dans {base.name}.")
|
||
|
|
return 0
|
||
|
|
manque = 0
|
||
|
|
for z in zones:
|
||
|
|
s = voute.get(nom_variable(z))
|
||
|
|
if not s:
|
||
|
|
print(f"{z} : AUCUNE CLE dans la voute ({nom_variable(z)}) — `dnssec.py generer {z}`.")
|
||
|
|
manque += 1
|
||
|
|
continue
|
||
|
|
print(f"{z}. IN DS {ds(z, s)}")
|
||
|
|
print(f" DNSKEY 257 3 13 {base64.b64encode(dnskey_rdata(s)[4:]).decode()}")
|
||
|
|
print("\nA remettre au registraire (type de condensat 2, SHA-256), APRES que les serveurs de "
|
||
|
|
"noms servent la zone signee — jamais avant.")
|
||
|
|
return 1 if manque else 0
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_verifier(args) -> int:
|
||
|
|
base = _instance()
|
||
|
|
voute = lire_voute(base)
|
||
|
|
code = 0
|
||
|
|
zones = zones_signees(base)
|
||
|
|
orphelines = sorted(k for k in voute if k.startswith("vault_dnssec_")
|
||
|
|
and k not in {nom_variable(z) for z in zones})
|
||
|
|
for z in zones:
|
||
|
|
s = voute.get(nom_variable(z))
|
||
|
|
if not s:
|
||
|
|
print(f"{z} : ECHEC — `dnssec: true` au plan, aucune cle dans la voute.")
|
||
|
|
code = 1
|
||
|
|
continue
|
||
|
|
attendu = ds(z, s).lower()
|
||
|
|
publies = ds_publies(z)
|
||
|
|
if publies is None:
|
||
|
|
print(f"{z} : MESURE IMPOSSIBLE — le DS publie n'a pu etre lu via {RESOLVEUR}.")
|
||
|
|
code = max(code, 2)
|
||
|
|
elif not publies:
|
||
|
|
print(f"{z} : cle presente ; le registre ne publie aucun DS (zone signee, pas encore ancree).")
|
||
|
|
elif attendu in publies:
|
||
|
|
print(f"{z} : CONFORME — le registre publie le DS de la cle de la voute.")
|
||
|
|
else:
|
||
|
|
print(f"{z} : ECHEC — le registre publie {sorted(publies)}, la voute donne {attendu}. "
|
||
|
|
"Tout resolveur validant tient ce domaine pour BOGUS.")
|
||
|
|
code = 1
|
||
|
|
for k in orphelines:
|
||
|
|
print(f"note : {k} est dans la voute, mais aucune zone `dnssec: true` ne la lit.")
|
||
|
|
return code
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_generer(args) -> int:
|
||
|
|
base = _instance()
|
||
|
|
z = args.zone.lower().rstrip(".")
|
||
|
|
reg = yaml.safe_load((base / "plan" / "domaines.yml").read_text(encoding="utf-8")) or {}
|
||
|
|
conf = (reg.get("domaines_publics") or {}).get(z)
|
||
|
|
if not conf or conf.get("autorite") != "primaire-cache":
|
||
|
|
raise SystemExit(f"REFUS : {z} n'est pas une zone `primaire-cache` de {base.name}.")
|
||
|
|
voute = lire_voute(base)
|
||
|
|
var = nom_variable(z)
|
||
|
|
if voute.get(var):
|
||
|
|
if not args.remplacer:
|
||
|
|
print(f"{z} : une cle existe deja ({var}). Rien n'est fait. DS : {ds(z, voute[var])}")
|
||
|
|
return 0
|
||
|
|
if os.environ.get("CONFIRMER") != "true":
|
||
|
|
raise SystemExit("REFUS : remplacer une cle DNSSEC exige CONFIRMER=true.")
|
||
|
|
publies = ds_publies(z)
|
||
|
|
if publies is None or publies:
|
||
|
|
raise SystemExit(
|
||
|
|
f"REFUS : le registre publie un DS pour {z} (ou la mesure a echoue). Remplacer la "
|
||
|
|
"cle maintenant rendrait le domaine BOGUS. Un roulement se fait en publiant "
|
||
|
|
"d'abord le nouveau DS a cote de l'ancien — hors de portee de cette commande.")
|
||
|
|
s = nouvelle_cle()
|
||
|
|
copie = ecrire_variable(base, var, s, f"DNSSEC : cle CSK ECDSA P-256 de {z} ({dt.date.today()})")
|
||
|
|
print(f"{z} : cle posee dans la voute de {base.name} ({var}), relue en place.")
|
||
|
|
print(f" DS : {ds(z, s)}")
|
||
|
|
print(f" copie de surete chiffree de l'ancienne voute : {copie}")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
|
|
sous = ap.add_subparsers(dest="commande", required=True)
|
||
|
|
p = sous.add_parser("ds", help="DS a remettre au registraire, depuis la voute")
|
||
|
|
p.add_argument("zone", nargs="?")
|
||
|
|
p.set_defaults(fn=cmd_ds)
|
||
|
|
p = sous.add_parser("verifier", help="voute, plan et registre concordent-ils ? (aucune ecriture)")
|
||
|
|
p.set_defaults(fn=cmd_verifier)
|
||
|
|
p = sous.add_parser("generer", help="cree la cle d'une zone dans la voute du locataire")
|
||
|
|
p.add_argument("zone")
|
||
|
|
p.add_argument("--remplacer", action="store_true")
|
||
|
|
p.set_defaults(fn=cmd_generer)
|
||
|
|
args = ap.parse_args()
|
||
|
|
return args.fn(args)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|