external
/
mailinabox
mirror of https://github.com/mail-in-a-box/mailinabox.git
1
0
Fork 0
Code Issues Releases Wiki Activity

When authoritative dns servers don't respond properly, retry 3 times

This commit is contained in:
Michael Kroes 2016-02-29 18:43:36 +01:00
parent 14f7ef6b20
commit cb94897e68
1 changed files with 35 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
management/status_checks.py
View File

@ -4,7 +4,7 @@
# TLS certificates have been signed, etc., and if not tells the user # TLS certificates have been signed, etc., and if not tells the user
# what to do next. # what to do next.
import sys, os, os.path, re, subprocess, datetime, multiprocessing.pool import sys, os, os.path, re, subprocess, datetime, multiprocessing.pool, time
import dns.reversename, dns.resolver import dns.reversename, dns.resolver
import dateutil.parser, dateutil.tz import dateutil.parser, dateutil.tz
@ -400,31 +400,21 @@ def check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles):
check_alias_exists("Hostmaster contact address", "hostmaster@" + domain, env, output) check_alias_exists("Hostmaster contact address", "hostmaster@" + domain, env, output)
def query_dns_ptr(qname): def query_dns_ptr(qname):
# Find the authoritative name server for the address using the default nameservers # When looking up PTR records bind will contact the authoritative servers for a response.
resolver = dns.resolver.get_default_resolver() # Sometimes these servers don't respond properly, we will give these servers 3 chances
nameserver = resolver.nameservers[0] # with a 2 second pause in between.
query = dns.message.make_query(qname, dns.rdatatype.PTR) for attempt in range(3):
timeout = 5 result=query_dns(qname, "PTR")
response = dns.query.udp(query, nameserver, timeout)
returnCode = response.rcode()
# Check that we were able to query the dns for the authoritative server # Check if the authoritative servers respond properly
if returnCode != dns.rcode.NOERROR: if result == "[nonameservers]":
return "[%s]" % dns.rcode.to_text(returnCode) time.sleep(2)
# If the current DNS server isn't the authority for this address use the one we find in the response
if len(response.authority) > 0:
rrset = response.authority[0]
else: else:
rrset = response.answer[0] # There might still be an error, like a timeout, but we will continue
# chances of recovering from those are slim.
break
rr = rrset[0] return result
if rr.rdtype != dns.rdatatype.SOA:
authority = rr.target
nameserver = query_dns(authority, "A")
# Resolve the PTR record using the proper name server
return query_dns(qname, "PTR", at=nameserver)
def check_alias_exists(alias_name, alias, env, output): def check_alias_exists(alias_name, alias, env, output):
mail_aliases = dict([(address, receivers) for address, receivers, *_ in get_mail_aliases(env)]) mail_aliases = dict([(address, receivers) for address, receivers, *_ in get_mail_aliases(env)])
@ -672,10 +662,12 @@ def query_dns(qname, rtype, nxdomain='[Not Set]', at=None):
# Do the query. # Do the query.
try: try:
response = resolver.query(qname, rtype) response = resolver.query(qname, rtype)
except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
# Host did not have an answer for this query; not sure what the # No response was received
# difference is between the two exceptions.
return nxdomain return nxdomain
except dns.resolver.NoNameservers:
# No non-broken nameservers were available to respond to the request
return "[nonameservers]"
except dns.exception.Timeout: except dns.exception.Timeout:
return "[timeout]" return "[timeout]"
@ -946,12 +938,21 @@ if __name__ == "__main__":
run_and_output_changes(env, pool) run_and_output_changes(env, pool)
elif sys.argv[1] == "--check-ptr": elif sys.argv[1] == "--check-ptr":
out = ConsoleOutput() # Run only the primary hostname checks, specifically the ptr check
# Get the list of domains we serve DNS zones for (i.e. does not include subdomains). shell('check_call', ["/usr/sbin/rndc", "flush"], trap=True)
dns_zonefiles = dict(get_dns_zones(env)) output = ConsoleOutput()
dns_domains = set(dns_zonefiles) domain=env["PRIMARY_HOSTNAME"]
check_primary_hostname_dns(env["PRIMARY_HOSTNAME"], env, out, dns_domains, dns_zonefiles) my_ips = env['PUBLIC_IP'] + ((" / "+env['PUBLIC_IPV6']) if env.get("PUBLIC_IPV6") else "")
existing_rdns_v4 = query_dns_ptr(dns.reversename.from_address(env['PUBLIC_IP']))
existing_rdns_v6 = query_dns_ptr(dns.reversename.from_address(env['PUBLIC_IPV6'])) if env.get("PUBLIC_IPV6") else None
if existing_rdns_v4 == domain and existing_rdns_v6 in (None, domain):
output.print_ok("Reverse DNS is set correctly at ISP. [%s ↦ %s]" % (my_ips, env['PRIMARY_HOSTNAME']))
elif existing_rdns_v4 == existing_rdns_v6 or existing_rdns_v6 is None:
output.print_error("""Your box's reverse DNS is currently %s, but it should be %s. Your ISP or cloud provider will have instructions
on setting up reverse DNS for your box.""" % (existing_rdns_v4, domain) )
else:
output.print_error("""Your box's reverse DNS is currently %s (IPv4) and %s (IPv6), but it should be %s. Your ISP or cloud provider will have instructions
on setting up reverse DNS for your box.""" % (existing_rdns_v4, existing_rdns_v6, domain) )
elif sys.argv[1] == "--check-primary-hostname": elif sys.argv[1] == "--check-primary-hostname":
# See if the primary hostname appears resolvable and has a signed certificate. # See if the primary hostname appears resolvable and has a signed certificate.