Daniel Allaire
712ad9f630
Cle CSK ECDSA P-256 tiree dans la voute du locataire, importee par le role, DS calcule sans la machine. Refus de changer la cle ou de retirer la signature tant qu'un DS est publie. SOA-EDIT EPOCH : INCEPTION-EPOCH aurait laisse expirer les signatures du site. CAA sur chezlepro.ca, sonde des signatures au site, P18 deplie vault_dnssec_<zone>. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
320 lines
14 KiB
Python
320 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())
|