ssl_certificates: reuse query_dns function in status_checks and simplify calls by...
authorJoshua Tauberer <jt@occams.info>
Sat, 12 May 2018 21:24:16 +0000 (17:24 -0400)
committerJoshua Tauberer <jt@occams.info>
Fri, 29 Jun 2018 20:46:21 +0000 (16:46 -0400)
management/ssl_certificates.py
management/status_checks.py

index c6b5080..303571b 100755 (executable)
@@ -213,41 +213,17 @@ def get_certificates_to_provision(env, show_extended_problems=True, force_domain
 
        # Filter out domains that we can't provision a certificate for.
        def can_provision_for_domain(domain):
-               from status_checks import normalize_ip
+               from status_checks import query_dns, normalize_ip
 
                # Does the domain resolve to this machine in public DNS? If not,
                # we can't do domain control validation. For IPv6 is configured,
                # make sure both IPv4 and IPv6 are correct because we don't know
                # how Let's Encrypt will connect.
-               import dns.resolver
                for rtype, value in [("A", env["PUBLIC_IP"]), ("AAAA", env.get("PUBLIC_IPV6"))]:
                        if not value: continue # IPv6 is not configured
-                       try:
-                               # Must make the qname absolute to prevent a fall-back lookup with a
-                               # search domain appended, by adding a period to the end.
-                               response = dns.resolver.query(domain + ".", rtype)
-                       except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer) as e:
-                               problems[domain] = "DNS isn't configured properly for this domain: DNS resolution failed (%s: %s)." % (rtype, str(e) or repr(e)) # NoAnswer's str is empty
-                               return False
-                       except Exception as e:
-                               problems[domain] = "DNS isn't configured properly for this domain: DNS lookup had an error: %s." % str(e)
-                               return False
-
-                       # Unfortunately, the response.__str__ returns bytes
-                       # instead of string, if it resulted from an AAAA-query.
-                       # We need to convert manually, until this is fixed:
-                       # https://github.com/rthalley/dnspython/issues/204
-                       #
-                       # BEGIN HOTFIX
-                       def rdata__str__(r):
-                               s = r.to_text()
-                               if isinstance(s, bytes):
-                                        s = s.decode('utf-8')
-                               return s
-                       # END HOTFIX
-
-                       if len(response) != 1 or normalize_ip(rdata__str__(response[0])) != normalize_ip(value):
-                               problems[domain] = "Domain control validation cannot be performed for this domain because DNS points the domain to another machine (%s %s)." % (rtype, ", ".join(rdata__str__(r) for r in response))
+                       response = query_dns(domain, rtype)
+                       if response != normalize_ip(value):
+                               problems[domain] = "The domain name does not resolve to this machine: DNS %s resolved to %s." % (rtype, response)
                                return False
 
                return True
index 6d07aee..3b0026d 100755 (executable)
@@ -393,7 +393,7 @@ def check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles):
 
        # Check that PRIMARY_HOSTNAME resolves to PUBLIC_IP[V6] in public DNS.
        ipv6 = query_dns(domain, "AAAA") if env.get("PUBLIC_IPV6") else None
-       if ip == env['PUBLIC_IP'] and not (ipv6 and env['PUBLIC_IPV6'] and normalize_ip(ipv6) != normalize_ip(env['PUBLIC_IPV6'])):
+       if ip == env['PUBLIC_IP'] and not (ipv6 and env['PUBLIC_IPV6'] and ipv6 != normalize_ip(env['PUBLIC_IPV6'])):
                output.print_ok("Domain resolves to box's IP address. [%s ↦ %s]" % (env['PRIMARY_HOSTNAME'], my_ips))
        else:
                output.print_error("""This domain must resolve to your box's IP address (%s) in public DNS but it currently resolves
@@ -640,7 +640,7 @@ def check_web_domain(domain, rounded_time, ssl_certificates, env, output):
                for (rtype, expected) in (("A", env['PUBLIC_IP']), ("AAAA", env.get('PUBLIC_IPV6'))):
                        if not expected: continue # IPv6 is not configured
                        value = query_dns(domain, rtype)
-                       if normalize_ip(value) == normalize_ip(expected):
+                       if value == normalize_ip(expected):
                                ok_values.append(value)
                        else:
                                output.print_error("""This domain should resolve to your box's IP address (%s %s) if you would like the box to serve
@@ -687,27 +687,17 @@ def query_dns(qname, rtype, nxdomain='[Not Set]', at=None):
        except dns.exception.Timeout:
                return "[timeout]"
 
+       # Normalize IP addresses. IP address --- especially IPv6 addresses --- can
+       # be expressed in equivalent string forms. Canonicalize the form before
+       # returning them. The caller should normalize any IP addresses the result
+       # of this method is compared with.
+       if rtype in ("A", "AAAA"):
+               response = [normalize_ip(str(r)) for r in response]
+
        # There may be multiple answers; concatenate the response. Remove trailing
        # periods from responses since that's how qnames are encoded in DNS but is
        # confusing for us. The order of the answers doesn't matter, so sort so we
        # can compare to a well known order.
-
-       # Unfortunately, the response.__str__ returns bytes
-       # instead of string, if it resulted from an AAAA-query.
-       # We need to convert manually, until this is fixed:
-       # https://github.com/rthalley/dnspython/issues/204
-       #
-       # BEGIN HOTFIX
-       response_new = []
-       for r in response:
-               s = r.to_text()
-               if isinstance(s, bytes):
-                       s = s.decode('utf-8')
-               response_new.append(s)
-       
-       response = response_new
-       # END HOTFIX
-
        return "; ".join(sorted(str(r).rstrip('.') for r in response))
 
 def check_ssl_cert(domain, rounded_time, ssl_certificates, env, output):
@@ -892,7 +882,9 @@ def run_and_output_changes(env, pool):
                json.dump(cur.buf, f, indent=True)
 
 def normalize_ip(ip):
-       # Use ipaddress module to normalize the IPv6 notation and ensure we are matching IPv6 addresses written in different representations according to rfc5952.
+       # Use ipaddress module to normalize the IPv6 notation and
+       # ensure we are matching IPv6 addresses written in different
+       # representations according to rfc5952.
        import ipaddress
        try:
                return str(ipaddress.ip_address(ip))