test: resolve the challenge through the authoritative nameservers

certbot reads challenge records straight from the zone's authoritative
servers, because a resolver that cached NXDOMAIN moments before the
record was written answers from that cache for the whole wait. It gets
there by asking NS for the zone and then A for each answer -- and the
mock served neither, so `authoritative_resolver` bailed on the first
question of every issuance and fell back to the system resolver.

The fallback is deliberate and issuance still worked, which is exactly
why nothing noticed the path was dead -- including this branch's own
change to it, generalizing the stripped label from `_acme-challenge` to
any challenge prefix, which had only unit coverage.

The mock now answers NS at a zone apex and A for the name it gives.
Only at the apex: a name below it must stay NODATA or the walk up
looking for the zone cut stops in the wrong place, and walking up is
what it does against a real zone. Both are gated on MOCK_CF_NS_ADDR, so
a mock that is not told where it can be reached advertises nothing.

The new assertion reads the mock's own query log, so it observes that
both questions were asked and answered rather than inferring it.

Author: kvinwang

dstack/gateway/test-run/e2e/docker-compose.yml

@@ -52,6 +52,11 @@ services:
       - DEBUG=true
       # The zones certbot writes into and Pebble reads back out of.
       - MOCK_CF_ZONES=test0.local,test1.local,test2.local,persist0.local,persist1.local,persist2.local,selfcheck0.local
+      # Advertise this container as the authoritative nameserver for those
+      # zones, so certbot's self-check can reach it the way it reaches a real
+      # one -- NS for the zone, A for the answer -- instead of failing that
+      # lookup and quietly falling back to its system resolver.
+      - MOCK_CF_NS_ADDR=172.30.0.10
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"]
       interval: 5s

dstack/gateway/test-run/e2e/test.sh

@@ -383,6 +383,22 @@ test_self_check_resolves_a_published_record() {
     [ "$(answered_queries_for "_validation-persist.${PERSIST_DOMAIN}")" -gt 0 ]
 }
 
+# The self-check does not ask its system resolver for the challenge record: a
+# resolver that cached NXDOMAIN moments before the record was written answers
+# from that cache for the whole wait, so the check reads the authoritative
+# servers instead. Getting there is two questions -- NS for the zone, then A for
+# the name that answers -- and if either goes unanswered certbot logs a warning
+# and silently falls back to the very resolver the path exists to avoid. The
+# fallback is deliberate and issuance still works, which is exactly why nothing
+# else here would notice that the path never ran.
+test_self_check_reads_the_authoritative_nameservers() {
+    # `_validation-persist.<zone>` is not a zone cut, so certbot strips the
+    # leading label and asks at the apex. That strip is what makes one code path
+    # serve both challenges, and this is where it is exercised for real.
+    [ "$(answered_queries_for "${PERSIST_DOMAIN}")" -gt 0 ] || return 1
+    [ "$(answered_queries_for "ns.${PERSIST_DOMAIN}")" -gt 0 ]
+}
+
 # The unhappy path, and the reason the check is advisory at all. A name with no
 # record has to be polled and given up on -- the record may still be
 # propagating -- rather than asked once and abandoned.
@@ -792,6 +808,8 @@ main() {
         "$(test_self_check_resolves_a_published_record; echo $?)"
     run_test "Self-check polls and gives up when the record is absent" \
         "$(test_self_check_gives_up_on_a_missing_record; echo $?)"
+    run_test "Self-check reads the authoritative nameservers" \
+        "$(test_self_check_reads_the_authoritative_nameservers; echo $?)"
 
     # Summary
     log_section "Test Summary"

tools/mock-cf-dns/server.py

@@ -45,6 +45,7 @@
 # under test resolved a name rather than inferring it from what happened next.
 QUERIES: list[dict[str, Any]] = []
 MAX_QUERIES = 2000
+QTYPE_ANY = 255
 NEXT_ID = 1
 # Names whose issuer CAA set went from non-empty to empty at some point.
 #
@@ -294,6 +295,46 @@ def txt_rdata(text: str) -> bytes:
     return b"".join(bytes([len(c)]) + c for c in chunks)
 
 
+def wire_name(name: str) -> bytes:
+    """Encode a domain name uncompressed, for use in RDATA."""
+    out = b""
+    for label in name.strip(".").split("."):
+        raw = label.encode("ascii", "ignore")[:63]
+        out += bytes([len(raw)]) + raw
+    return out + b"\x00"
+
+
+def ns_address() -> str | None:
+    """Address to advertise as this mock's nameserver, if it is to advertise one.
+
+    A client that reads a challenge record straight from the authoritative
+    servers -- which is the only way to dodge a recursive resolver's negative
+    cache -- gets there by asking NS for the zone and then A for each answer.
+    Unset, this mock answers neither and that client falls back to its system
+    resolver, so the authoritative path never runs. Set to the address this
+    container is reachable at, and the path is exercised end to end.
+    """
+    return os.environ.get("MOCK_CF_NS_ADDR") or None
+
+
+def _synthesized(name: str, qtype: int) -> list[tuple[int, bytes]]:
+    """NS at a zone apex and A for the nameserver it names.
+
+    Only at the apex: a name below it must answer NODATA, or a client walking
+    up from `_acme-challenge.<name>` looking for the zone cut would stop at the
+    wrong place -- and walking up is what it does against a real zone.
+    """
+    addr = ns_address()
+    if not addr:
+        return []
+    apexes = {zone["name"] for zone in _zones()}
+    if qtype in (2, QTYPE_ANY) and name in apexes:
+        return [(2, wire_name(f"ns.{name}"))]
+    if qtype in (1, QTYPE_ANY) and name.startswith("ns.") and name[3:] in apexes:
+        return [(1, socket.inet_aton(addr))]
+    return []
+
+
 def _upstream() -> str:
     """Where to send questions this mock is not authoritative for.
 
@@ -371,21 +412,23 @@ def dns_response(packet: bytes) -> bytes:
             return txid + struct.pack("!HHHHH", 0x8182, 1, 0, 0, 0) + packet[12:qend + 4]
 
     with STATE_LOCK:
-        answers = [
+        stored = [
             r
             for r in RECORDS
             if r["type"] == "TXT" and r["name"].strip(".").lower() == name
         ]
-    if qtype not in (16, 255):
-        answers = []
+    if qtype not in (16, QTYPE_ANY):
+        stored = []
+    answers = [(16, txt_rdata(r["content"])) for r in stored]
+    answers += _synthesized(name, qtype)
+
     _record_query(name, qtype, len(answers), False)
     header = txid + struct.pack("!HHHHH", 0x8180, 1, len(answers), 0, 0)
     body = question
-    for record in answers:
-        rdata = txt_rdata(record["content"])
+    for rtype, rdata in answers:
         body += (
             b"\xc0\x0c"
-            + struct.pack("!HHIH", 16, 1, int(record.get("ttl", 60)), len(rdata))
+            + struct.pack("!HHIH", rtype, 1, 60, len(rdata))
             + rdata
         )
     return header + body