fix(dns): query correct NS if MNAME server is hidden

.github/workflows/ci.yaml

@@ -57,8 +57,9 @@ jobs:
 
   lxc-test:
     name: LXC deploy and test
-    uses: chatmail/cmlxc/.github/workflows/lxc-test.yml@v0.10.0
+    uses: chatmail/cmlxc/.github/workflows/lxc-test.yml@v0.13.5
     with:
+      cmlxc_version: v0.13.5
       cmlxc_commands: |
         cmlxc init 
         # single cmdeploy relay test

cmdeploy/src/cmdeploy/remote/rdns.py

@@ -64,21 +64,25 @@ def get_dkim_entry(mail_domain, pre_command, dkim_selector):
     )
 
 
-def query_dns(typ, domain):
-    # Get autoritative nameserver from the SOA record.
-    soa_answers = [
+def get_authoritative_ns(domain):
+    ns_replies = [
         x.split()
         for x in shell(
-            f"dig -r -q {domain} -t SOA +noall +authority +answer", print=log_progress
+            f"dig -r -q {domain} -t NS +noall +authority +answer", print=log_progress
         ).split("\n")
     ]
-    soa = [a for a in soa_answers if len(a) >= 3 and a[3] == "SOA"]
-    if not soa:
+    filtered_replies = [a for a in ns_replies if len(a) >= 5 and a[3] == "NS"]
+    if not filtered_replies:
         return
-    ns = soa[0][4]
+    return filtered_replies[0][4]
+
+
+def query_dns(typ, domain):
+    ns = get_authoritative_ns(domain)
 
     # Query authoritative nameserver directly to bypass DNS cache.
-    res = shell(f"dig @{ns} -r -q {domain} -t {typ} +short", print=log_progress)
+    direct_ns = f"@{ns}" if ns else ""
+    res = shell(f"dig {direct_ns} -r -q {domain} -t {typ} +short", print=log_progress)
     return next((line for line in res.split("\n") if not line.startswith(";")), "")
 
 

cmdeploy/src/cmdeploy/tests/test_dns.py

@@ -4,6 +4,7 @@
 
 from cmdeploy import remote
 from cmdeploy.dns import check_full_zone, check_initial_remote_data, parse_zone_records
+from cmdeploy.remote.rdns import get_authoritative_ns
 
 
 @pytest.fixture
@@ -14,11 +15,15 @@ def shell(command, fail_ok=False, print=print):
         if command.startswith("dig"):
             if command == "dig":
                 return "."
-            if "SOA" in command:
+            if "with.public.soa" in command and "NS" in command:
+                return "domain.with.public.soa. 2419 IN NS ns1.first-ns.de."
+            if "with.hidden.soa" in command and "NS" in command:
                 return (
-                    "delta.chat. 21600 IN SOA ns1.first-ns.de. dns.hetzner.com."
-                    " 2025102800 14400 1800 604800 3600"
+                    "domain.with.hidden.soa. 2137 IN NS ns1.desec.io.\n"
+                    "domain.with.hidden.soa. 2137 IN NS ns2.desec.org."
                 )
+            if "NS" in command:
+                return "delta.chat. 21600 IN NS ns1.first-ns.de."
             command_chunks = command.split()
             domain, typ = command_chunks[4], command_chunks[6]
             try:
@@ -125,6 +130,17 @@ def test_perform_initial_checks_no_mta_sts_self_signed(self, mockdns):
         assert not l
 
 
+@pytest.mark.parametrize(
+    ("domain", "ns"),
+    [
+        ("domain.with.public.soa", "ns1.first-ns.de."),
+        ("domain.with.hidden.soa", "ns1.desec.io."),
+    ],
+)
+def test_get_authoritative_ns(domain, ns, mockdns):
+    assert get_authoritative_ns(domain) == ns
+
+
 def test_parse_zone_records():
     text = """
     ; This is a comment