From efdaefc39e930994a8c394c58d12fb554b820cb8 Mon Sep 17 00:00:00 2001 From: duffyduck Date: Thu, 13 Aug 2026 10:21:30 +0200 Subject: [PATCH] Nameserver ohne Route ueberspringen statt abzustuerzen Die Nameserver von fon-aria.de haben AAAA-Records, der Container hat aber kein IPv6. dns.query.udp() ist deshalb mit OSError [Errno 99] "Cannot assign requested address" durchgeschlagen und hat den ganzen Lauf abgebrochen - nach dem Anlegen der Challenge-Records. Adressen, zu denen es keine Route gibt, werden jetzt vorher aussortiert (UDP-connect, verschickt nichts) und waehrend des Wartens zusaetzlich verworfen, falls doch eine ausfaellt. Bleibt keine uebrig, gibt es eine klare Meldung statt eines Tracebacks. Bei fon-aria.de bleiben so die beiden IPv4-Adressen derselben Nameserver uebrig. - OSError wird in main() sauber abgefangen - tests/test_dns.py deckt die Faelle ab; im Container gegen die echte Zone gegengeprueft Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 2 ++ app/__init__.py | 2 +- app/dnsutil.py | 67 ++++++++++++++++++++++++++++-------- app/main.py | 3 ++ tests/test_dns.py | 86 +++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 145 insertions(+), 15 deletions(-) create mode 100644 tests/test_dns.py diff --git a/README.md b/README.md index d24f777..e5032ed 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,7 @@ Das Skript ist idempotent: Läuft es, obwohl das Zertifikat noch länger als | `Plesk rejected the credentials (HTTP 401)` | API-Key falsch oder nicht für diese Quell-IP erzeugt | | `No Plesk DNS zone found for …` | Zone liegt nicht auf diesem Plesk oder der API-User sieht sie nicht. Die Meldung listet auf, welche Namen probiert wurden und was Plesk dazu gesagt hat. Notfalls die Zone mit `--zone example.com` fest vorgeben | | `DNS propagation timed out` | Plesk ist nicht der autoritative NS, oder die Zone ist deaktiviert. Mit `-v` prüfen, welcher Nameserver antwortet | +| `Skipping unreachable nameserver address(es): 2001:…` | Normal: der Container hat kein IPv6, die IPv4-Adressen derselben Nameserver werden benutzt | | `no dns-01 challenge` | Wildcards gehen ausschließlich über DNS-01 – Domain-Validierung per HTTP ist nicht möglich | | `urn:ietf:params:acme:error:rateLimited` | Rate-Limit von Let's Encrypt (5 Zertifikate pro Woche pro Domain). Erst mit `--staging` testen | | PFX lässt sich in Windows nicht importieren | `PFX_LEGACY_COMPAT=true` setzen und neu erzeugen | @@ -178,5 +179,6 @@ Ohne echten Plesk-Server und ohne Zertifikatsausstellung: ```bash .venv/bin/python tests/test_offline.py # XML-Parsing, DNS-Logik, Dateiausgabe, PFX .venv/bin/python tests/test_cli.py # kompletter Ablauf gegen ein Plesk-Fake +.venv/bin/python tests/test_dns.py # Propagations-Check, unerreichbare Nameserver .venv/bin/python tests/test_acme_api.py # prueft die Let's-Encrypt-Anbindung (nur Directory-Abruf) ``` diff --git a/app/__init__.py b/app/__init__.py index 8c93f6f..996a9fd 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1,3 +1,3 @@ """Wildcard Let's Encrypt certificate creator for Plesk-managed DNS zones.""" -__version__ = "1.1.0" +__version__ = "1.1.1" diff --git a/app/dnsutil.py b/app/dnsutil.py index 9bba3cf..cfe8051 100644 --- a/app/dnsutil.py +++ b/app/dnsutil.py @@ -3,6 +3,7 @@ from __future__ import annotations import logging +import socket import time import dns.exception @@ -36,7 +37,7 @@ def authoritative_servers(zone: str, fallback_resolvers: list[str]) -> list[str] names = sorted({str(rdata.target).rstrip(".") for rdata in answer}) if names: break - except dns.exception.DNSException as exc: + except (dns.exception.DNSException, OSError) as exc: log.debug("NS lookup for %s failed: %s", zone, exc) lookup = _resolver(fallback_resolvers) @@ -45,19 +46,41 @@ def authoritative_servers(zone: str, fallback_resolvers: list[str]) -> list[str] try: for rdata in lookup.resolve(name, rtype, raise_on_no_answer=False): ips.append(str(rdata)) - except dns.exception.DNSException: + except (dns.exception.DNSException, OSError): continue unique = list(dict.fromkeys(ips)) - if unique: - log.info("Authoritative nameservers for %s: %s (%s)", zone, ", ".join(names), ", ".join(unique)) + usable = [ip for ip in unique if reachable(ip)] + dropped = [ip for ip in unique if ip not in usable] + if dropped: + # Typical inside a container without IPv6: the NS have AAAA records we + # cannot talk to. Their IPv4 addresses answer the same zone data. + log.info("Skipping unreachable nameserver address(es): %s", ", ".join(dropped)) + if usable: + log.info("Authoritative nameservers for %s: %s (%s)", zone, ", ".join(names), ", ".join(usable)) else: - log.warning("Could not determine authoritative nameservers for %s, using public resolvers.", zone) - return unique + log.warning("Could not reach any authoritative nameserver for %s, using public resolvers.", zone) + return usable -def txt_values(name: str, server: str) -> set[str]: - """TXT values for *name* as seen by the DNS server at *server*.""" +def reachable(server: str) -> bool: + """Can this machine send a UDP packet to *server* at all? + + ``connect()`` on a UDP socket sends nothing, it only resolves the route - + exactly what we need to weed out IPv6 addresses on an IPv4-only host. + """ + family = socket.AF_INET6 if ":" in server else socket.AF_INET + try: + with socket.socket(family, socket.SOCK_DGRAM) as sock: + sock.connect((server, 53)) + return True + except OSError as exc: + log.debug("Nameserver %s is not reachable from here: %s", server, exc) + return False + + +def txt_values(name: str, server: str) -> set[str] | None: + """TXT values for *name* as seen by *server*. None if the server is unusable.""" query = dns.message.make_query(name, dns.rdatatype.TXT) values: set[str] = set() try: @@ -67,6 +90,10 @@ def txt_values(name: str, server: str) -> set[str]: except dns.exception.DNSException as exc: log.debug("TXT query %s @%s failed: %s", name, server, exc) return values + except OSError as exc: + # No route to that address (e.g. IPv6 without IPv6 connectivity). + log.debug("TXT query %s @%s not possible: %s", name, server, exc) + return None for rrset in response.answer: if rrset.rdtype != dns.rdatatype.TXT: @@ -89,19 +116,31 @@ def wait_for_txt( return False deadline = time.monotonic() + timeout + alive = list(servers) attempt = 0 while True: attempt += 1 missing: list[str] = [] + unusable: list[str] = [] for name, wanted in expected.items(): - for server in servers: + for server in alive: seen = txt_values(name, server) - for value in wanted: - if value not in seen: - missing.append(f"{name} @{server}") - break + if seen is None: + unusable.append(server) + elif not wanted <= seen: + missing.append(f"{name} @{server}") + + if unusable: + for server in dict.fromkeys(unusable): + alive.remove(server) + log.warning("Nameserver %s cannot be queried from here - ignoring it.", server) + if not alive: + log.error("None of the nameservers could be queried at all.") + return False + continue # judge the round again, now without the dead servers + if not missing: - log.info("DNS propagation confirmed on all %d nameserver(s).", len(servers)) + log.info("DNS propagation confirmed on all %d nameserver(s).", len(alive)) return True remaining = deadline - time.monotonic() diff --git a/app/main.py b/app/main.py index 589e325..a7e1be2 100644 --- a/app/main.py +++ b/app/main.py @@ -225,6 +225,9 @@ def main(argv: list[str] | None = None) -> int: except AcmeFailure as exc: log.error("%s", exc) return 1 + except OSError as exc: + log.error("Network error: %s", exc) + return 1 except KeyboardInterrupt: # pragma: no cover log.error("Aborted.") return 130 diff --git a/tests/test_dns.py b/tests/test_dns.py new file mode 100644 index 0000000..06ced87 --- /dev/null +++ b/tests/test_dns.py @@ -0,0 +1,86 @@ +"""DNS-Propagationslogik: nicht erreichbare Nameserver (z.B. IPv6 ohne IPv6) fliegen raus.""" +import pathlib +import sys + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +from app import dnsutil + +V4A, V4B, V6 = "37.157.254.21", "5.199.143.173", "2001:4ba0:cafe:918::1" +NAME = "_acme-challenge.endian.fon-aria.de" +WANTED = {"value-a", "value-b"} + +dnsutil.time.sleep = lambda _s: None # keine echten Wartezeiten im Test + +# ---- reachable() ---------------------------------------------------------- +assert dnsutil.reachable("127.0.0.1") is True +assert dnsutil.reachable("not-an-ip") is False +print("reachable ok") + +# ---- authoritative_servers() filtert unerreichbare Adressen --------------- +dnsutil.reachable = lambda ip: ":" not in ip # so tut ein Host ohne IPv6 +resolved = {"ns1.fon-aria.de": [V4A, V6], "ns2.fon-aria.de": [V4B]} + + +class FakeAnswer(list): + pass + + +class FakeResolver: + def __init__(self, *a, **kw): + self.nameservers = [] + self.lifetime = self.timeout = 5 + + def resolve(self, qname, rtype, raise_on_no_answer=False): + qname = str(qname).rstrip(".") + if rtype == "NS": + return FakeAnswer([type("R", (), {"target": n + "."})() for n in resolved]) + if rtype == "A": + return FakeAnswer([ip for ip in resolved.get(qname, []) if ":" not in ip]) + if rtype == "AAAA": + return FakeAnswer([ip for ip in resolved.get(qname, []) if ":" in ip]) + return FakeAnswer() + + +dnsutil.dns.resolver.Resolver = FakeResolver +servers = dnsutil.authoritative_servers("fon-aria.de", ["1.1.1.1"]) +assert servers == [V4A, V4B], servers +print("authoritative_servers ok:", servers) + +# ---- wait_for_txt: der tote Server darf das Ergebnis nicht blockieren ----- +calls = [] + + +def fake_txt(name, server): + calls.append(server) + if server == V6: + return None # kein Weg dorthin + return set(WANTED) # beide Werte da + + +dnsutil.txt_values = fake_txt +assert dnsutil.wait_for_txt({NAME: WANTED}, [V4A, V6, V4B], timeout=30, interval=1) is True +assert V6 in calls and calls.count(V6) == 1, calls # genau einmal probiert, dann verworfen +print("wait_for_txt ignoriert toten Nameserver ok") + +# alle tot -> sauberes False statt Absturz +dnsutil.txt_values = lambda name, server: None +assert dnsutil.wait_for_txt({NAME: WANTED}, [V6], timeout=30, interval=1) is False + +# Wert fehlt auf einem Server -> Timeout, kein Haenger +state = {"n": 0} + + +def slowly(name, server): + state["n"] += 1 + return set(WANTED) if state["n"] > 3 else {"value-a"} + + +dnsutil.txt_values = slowly +assert dnsutil.wait_for_txt({NAME: WANTED}, [V4A], timeout=30, interval=1) is True + +dnsutil.txt_values = lambda name, server: {"value-a"} +assert dnsutil.wait_for_txt({NAME: WANTED}, [V4A], timeout=2, interval=1) is False +print("wait_for_txt Timeout-Pfad ok") + +print("\nALLE DNS-TESTS BESTANDEN")