ssl_certificates: reuse query_dns function in status_checks and simplify calls by calling normalize_ip within query_dns
This commit is contained in:
parent
f9a0e39cc9
commit
8be23d5ef6
|
@ -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
|
||||
|
|
|
@ -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))
|
||||
|
|
Loading…
Reference in New Issue