cmdeploy: only output DNS entries which are not correct yet

cmdeploy/src/cmdeploy/chatmail.zone.f

@@ -6,7 +6,7 @@ _imaps._tcp.{chatmail_domain}.       SRV 0 1 993 {chatmail_domain}.
 {chatmail_domain}.                IN CAA 128 issue "letsencrypt.org;accounturi={acme_account_url}"
 {chatmail_domain}.                   TXT "v=spf1 a:{chatmail_domain} -all"
 _dmarc.{chatmail_domain}.            TXT "v=DMARC1;p=reject;rua=mailto:{email};ruf=mailto:{email};fo=1;adkim=r;aspf=r"
-_mta-sts.{chatmail_domain}.       IN TXT "v=STSv1; id={sts_id}"
+_mta-sts.{chatmail_domain}.          TXT "v=STSv1; id={sts_id}"
 mta-sts.{chatmail_domain}.        IN CNAME {chatmail_domain}.
-_smtp._tls.{chatmail_domain}.     IN TXT "v=TLSRPTv1;rua=mailto:{email}"
+_smtp._tls.{chatmail_domain}.        TXT "v=TLSRPTv1;rua=mailto:{email}"
 {dkim_entry}

cmdeploy/src/cmdeploy/cmdeploy.py

@@ -15,7 +15,7 @@ from pathlib import Path
 
 from termcolor import colored
 from chatmaild.config import read_config, write_initial_config
-from cmdeploy.dns import resolve
+from cmdeploy.dns import resolve, resolve_mx, get
 
 
 #
@@ -104,24 +104,62 @@ def dns_cmd(args, out):
             lines.append(line)
         return "\n".join(lines)
 
+    print("Checking your DKIM keys and DNS entries...")
     acme_account_url = out.shell_output(f"{ssh} -- acmetool account-url")
     dkim_entry = read_dkim_entries(out.shell_output(f"{ssh} -- opendkim-genzone -F"))
 
-    out(
+    to_print = []
-        f"[writing {args.config.mail_domain} zone data (using space as separator) to stdout output]",
+    with open(template, "r") as f:
-        green=True,
+        for line in f.readlines():
-    )
+            line = line.format(
-    print(
+                acme_account_url=acme_account_url,
-        template.read_text()
+                email=f"root@{args.config.mail_domain}",
-        .format(
+                sts_id=datetime.datetime.now().strftime("%Y%m%d%H%M"),
-            acme_account_url=acme_account_url,
+                chatmail_domain=args.config.mail_domain,
-            email=f"root@{args.config.mail_domain}",
+                dkim_entry=dkim_entry,
-            sts_id=datetime.datetime.now().strftime("%Y%m%d%H%M"),
+            ).strip()
-            chatmail_domain=args.config.mail_domain,
+            if " MX " in line:
-            dkim_entry=dkim_entry,
+                domain, typ, prio, value = line.split()
+                current = resolve_mx(domain[:-1])
+                if not current[0]:
+                    print(line)
+                elif current[1] != value:
+                    print(line.replace(prio, str(current[0] + 1)))
+            if " SRV " in line:
+                domain, typ, prio, weight, port, value = line.split()
+                current = get("SRV", domain[:-1])
+                if current != f"{prio} {weight} {port} {value}":
+                    print(line)
+            if " CAA " in line:
+                domain, value = line.split(" IN CAA ")
+                current = get("CAA", domain.strip()[:-1])
+                if current != value:
+                    print(line)
+            if "  TXT " in line:
+                domain, value = line.split(" TXT ")
+                current = get("TXT", domain.strip()[:-1])
+                if domain.startswith("_mta-sts."):
+                    if current.split("id=")[0] == value.split("id=")[0]:
+                        continue
+                if current != value:
+                    print(line)
+            if " IN TXT ( " in line:
+                line += f.read()
+                domain, data = line.split(" IN TXT ")
+                current = get("TXT", domain.strip()[:-1]).replace('" "', '"\n "')
+                current = f"( {current} )"
+                if current.replace(";", "\\;") != data:
+                    print(
+                        "wrong:   '", current.replace(";", "\\;"), "'"
+                    )
+                    print("missing: '", data, "'")
+    if to_print:
+        to_print.insert(
+            0, "\nYou should configure the following DNS entries at your provider:\n"
         )
-        .strip()
+        print("\n".join(to_print))
-    )
+    else:
+        out.green("Great! All your DNS entries are correct.")
 
 
 def status_cmd(args, out):

cmdeploy/src/cmdeploy/dns.py

@@ -1,26 +1,19 @@
 import requests
 
+url = "https://dns.nextdns.io/dns-query"
 dns_types = {
     "A": 1,
     "AAAA": 28,
     "CNAME": 5,
+    "MX": 15,
+    "SRV": 33,
+    "CAA": 257,
+    "TXT": 16,
 }
 
 
-def resolve(domain: str) -> str:
-    result = get("A", domain)
-    if not result:
-        result = get("CNAME", domain)
-        if result:
-            result = get("A", result[:-1])
-            if not result:
-                result = get("AAAA", domain)
-    return result
-
-
 def get(typ: str, domain: str) -> str:
     """Get a DNS entry"""
-    url = "https://dns.nextdns.io/dns-query"
     r = requests.get(
         url,
         params={"name": domain, "type": typ},
@@ -32,3 +25,35 @@ def get(typ: str, domain: str) -> str:
         for answer in j["Answer"]:
             if answer["type"] == dns_types[typ]:
                 return answer["data"]
+    return ""
+
+
+def resolve_mx(domain: str) -> (str, str):
+    """Resolve an MX entry"""
+    r = requests.get(
+        url,
+        params={"name": domain, "type": "MX"},
+        headers={"accept": "application/dns-json"},
+    )
+
+    j = r.json()
+    if "Answer" in j:
+        result = (0, None)
+        for answer in j["Answer"]:
+            if answer["type"] == dns_types["MX"]:
+                prio, server_name = answer["data"].split()
+                if int(prio) > result[0]:
+                    result = (int(prio), server_name)
+        return result
+    return None, None
+
+
+def resolve(domain: str) -> str:
+    result = get("A", domain)
+    if not result:
+        result = get("CNAME", domain)
+        if result:
+            result = get("A", result[:-1])
+            if not result:
+                result = get("AAAA", domain)
+    return result