13

As part of a tool I'm writing I want to have a diagnostic that will tell the user whether they have configured their domain's DNS correctly for a particular service. I want to query the authoritative DNS server for their domain so that I can bypass any cached results.

Jon Colverson
  • 2,986
  • 1
  • 24
  • 24

5 Answers5

19

Here's my attempt at this. It uses the system's standard DNS server for looking up the root server for the top-level domain and for resolving the names of the various DNS servers along the chain, which I think is appropriate because those names would presumably change very infrequently.

import dns
import dns.name
import dns.query
import dns.resolver

def get_authoritative_nameserver(domain, log=lambda msg: None):
    n = dns.name.from_text(domain)

    depth = 2
    default = dns.resolver.get_default_resolver()
    nameserver = default.nameservers[0]

    last = False
    while not last:
        s = n.split(depth)

        last = s[0].to_unicode() == u'@'
        sub = s[1]

        log('Looking up %s on %s' % (sub, nameserver))
        query = dns.message.make_query(sub, dns.rdatatype.NS)
        response = dns.query.udp(query, nameserver)

        rcode = response.rcode()
        if rcode != dns.rcode.NOERROR:
            if rcode == dns.rcode.NXDOMAIN:
                raise Exception('%s does not exist.' % sub)
            else:
                raise Exception('Error %s' % dns.rcode.to_text(rcode))

        rrset = None
        if len(response.authority) > 0:
            rrset = response.authority[0]
        else:
            rrset = response.answer[0]

        rr = rrset[0]
        if rr.rdtype == dns.rdatatype.SOA:
            log('Same server is authoritative for %s' % sub)
        else:
            authority = rr.target
            log('%s is authoritative for %s' % (authority, sub))
            nameserver = default.query(authority).rrset[0].to_text()

        depth += 1

    return nameserver


import sys

def log(msg):
    print msg

print get_authoritative_nameserver(sys.argv[1], log)

Here's some sample output:

Looking up com. on 192.168.255.10
l.gtld-servers.net. is authoritative for com.
Looking up stackoverflow.com. on 192.41.162.30
ns1.p19.dynect.net. is authoritative for stackoverflow.com.
Looking up meta.stackoverflow.com. on 208.78.70.19
Same server is authoritative for meta.stackoverflow.com.
208.78.70.19
Jon Colverson
  • 2,986
  • 1
  • 24
  • 24
8

I came across Jon Colverson's answer, and it helped me understand the dnspython module and how to process the results (I guess all DNS modules have the same twisty maze of class structure ...) I needed the TTL and the glue records, so I created my own adaptation. I am posting it here in case somebody would find it useful; I do not intend to compete with Jon Colverson's excellent answer, just fill in some additional blanks. The basic improvement is the use of name server information from the additional section of the answer, where available. I suppose a server could put something else than glue records in the additional section, so perhaps this should still be enhanced to properly correlate the information from the additional section with the information in the answer section. I also fetch and print all the name servers, not just the first one.

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import dns.query
import dns.resolver
from dns.exception import DNSException

def query_authoritative_ns (domain, log=lambda msg: None):

    default = dns.resolver.get_default_resolver()
    ns = default.nameservers[0]

    n = domain.split('.')

    for i in xrange(len(n), 0, -1):
        sub = '.'.join(n[i-1:])

        log('Looking up %s on %s' % (sub, ns))
        query = dns.message.make_query(sub, dns.rdatatype.NS)
        response = dns.query.udp(query, ns)

        rcode = response.rcode()
        if rcode != dns.rcode.NOERROR:
            if rcode == dns.rcode.NXDOMAIN:
                raise Exception('%s does not exist.' % (sub))
            else:
                raise Exception('Error %s' % (dns.rcode.to_text(rcode)))

        if len(response.authority) > 0:
            rrsets = response.authority
        elif len(response.additional) > 0:
            rrsets = [response.additional]
        else:
            rrsets = response.answer

        # Handle all RRsets, not just the first one
        for rrset in rrsets:
            for rr in rrset:
                if rr.rdtype == dns.rdatatype.SOA:
                    log('Same server is authoritative for %s' % (sub))
                elif rr.rdtype == dns.rdatatype.A:
                    ns = rr.items[0].address
                    log('Glue record for %s: %s' % (rr.name, ns))
                elif rr.rdtype == dns.rdatatype.NS:
                    authority = rr.target
                    ns = default.query(authority).rrset[0].to_text()
                    log('%s [%s] is authoritative for %s; ttl %i' % 
                        (authority, ns, sub, rrset.ttl))
                    result = rrset
                else:
                    # IPv6 glue records etc
                    #log('Ignoring %s' % (rr))
                    pass

    return result

import sys

def log (msg):
    sys.stderr.write(msg + u'\n')

for s in sys.argv[1:]:
    print query_authoritative_ns (s, log)
tripleee
  • 175,061
  • 34
  • 275
  • 318
2

The other examples are fine but overly complex if you need just the nameservers. Example from http://c0deman.wordpress.com/2014/06/17/find-nameservers-of-domain-name-python/ :

import dns.resolver

domain = 'google.com'
answers = dns.resolver.query(domain,'NS')
for server in answers:
    print server
Willem
  • 3,043
  • 2
  • 25
  • 37
  • 2
    Would I be right in thinking that might return cached results, though? In my case I specifically wanted to find the current server avoiding any caching, but there might be a significantly simpler way of doing that than the way I did it. :) – Jon Colverson Jul 02 '14 at 18:43
  • 3
    That will only give you the nameserver for the top-level domain (or sub-domains if they have NS records). It won't tell you what the authoritative DNS server is for `www.example.org` and will raise a `dns.resolver.NoAnswer` exception. – pwaring Jun 10 '16 at 15:29
1

Im pretty sure this would do it.

import dns.resolver

domain = 'co.uk'

response = dns.resolver.query(domain, 'SOA')
if response.rrset is not None:
    print response.rrset

You could of course cleanup the response

import dns.resolver
import re

domain = 'co.uk'

response = dns.resolver.query(domain, 'SOA')

if response.rrset is not None:
    pattern= r'(%s)\.\s(\d{1,})\s(\w+)\sSOA\s(.*?)\.\s(.*?)\.\s(\d{1,})\s(\d{1,})\s(\d{1,})\s(\d{1,})\s(\d{1,})' % domain
    match = re.match(pattern, str(response.rrset))
    m_name, ttl, class_, ns, email, serial, refresh, retry, expiry, minim = match.groups()

output ='''
Main Name In Zone: {a},
Cache TTL: {b},
Class: {c},
Authoritive NS: {d},
Email Address: {e},
Last Change: {f},
Retry In Secs: {g},
Expiry: {h},
Slave Cache In Sec: {i}
'''.format(a = m_name, b = ttl, c = class_, d = ns, e = str(email).replace('\\', ''), f = serial, g = retry, h = expiry, i = minim)

print output

This produces

Main Name In Zone: co.uk,
Cache TTL: 600,
Class: IN,
Authoritive NS: dns1.nic.uk,
Email Address: hostmaster.nominet.org.uk,
Last Change: 1305857394,
Retry In Secs: 300,
Expiry: 2419200,
Slave Cache In Sec: 10800
iNoob
  • 1,375
  • 3
  • 19
  • 47
  • 2
    Correct me if I'm wrong, but I believe those SOA results could still be cached on the resolver, though? It was bypassing the caching that made things more elaborate. – Jon Colverson Aug 25 '17 at 23:40
1

I ended up here because I needed to get an accurate return of ns in python ignoring the cache and starting from random servers.

I got a lot of help from your answers to build what is for me, the most useful version.

It allows me to know directly if a site belongs to a domain parking or not (e.g. freespt.com => ['ns1.bodis.com.', 'ns2.bodis.com.'])

I hope my code will be useful to you!

import random
import sys

import dns.resolver
import dns.name
import dns.message
import dns.query
import dns.flags


import re

NAMESERVERS = {
    '1.1.1.1': 'Cloudflare DNS',
    '1.0.0.1': 'Cloudflare DNS',
    '8.8.8.8': 'Google DNS',
    '8.8.4.4': 'Google DNS',
    '9.9.9.9': 'Quad9',
    '149.112.112.112': 'Quad9',
    '208.67.222.222': 'OpenDNS',
    '208.67.220.220': 'OpenDNS',
    '185.228.168.9': 'CleanBrowsing',
    '185.228.169.9': 'CleanBrowsing',
    '94.140.14.14': 'AdGuard DNS',
    '94.140.15.15': 'AdGuard DNS',
}

def rand_nameserver():
    """
    Choosing randomly one of the nameservers
    """
    return random.choice(list(NAMESERVERS.keys()))

def query_authoritative_ns(domain):
    """
    Dig recursively leaf to root and retrieve ns with cache bypassing.
    That will NOT only give you the nameserver for the top-level domain !
    It'll try sub-domains if they have NS records.

    It'll tell you what the authoritative DNS server is for www.example.org and will not raise a dns.resolver.NoAnswer exception.

    (quite similary to dig +trace)
    """

    trace = '### Querying authoritative ns for {d}:\n'.format(d=domain)
    result = None

    resolver = dns.resolver.Resolver(configure=False)
    name_server = rand_nameserver()

    # Setting-up random nameservers
    resolver.nameservers = [name_server]

    # Defining Timeout and lifetime
    # To not taking long time to skip to the next one records when it failed.
    resolver.timeout = 5
    resolver.lifetime = 5

    trace += '[NS records] [{d}] Used nameserver for DNS NS query is {fns} ({ns})'.format(
        d=domain,
        fns=NAMESERVERS.get(name_server),
        ns=name_server
    )

    ns = resolver.nameservers[0]

    # branches
    n = domain.split('.')

    for i in range(len(n), 0, -1):
        sub = '.'.join(n[i - 1:])

        trace += '\n[NS records] Looking up %s on %s' % (sub, ns)
        query = dns.message.make_query(sub, dns.rdatatype.NS)
        try:
            response = dns.query.udp(query, ns, port=53, timeout=5)

        except Exception as e:

            trace += '\n[NS records] [domain={d}] [subdomain={s}] Receive exception : {e}'.format(d=domain, s=sub, e=e)
            continue

        rcode = response.rcode()
        if rcode != dns.rcode.NOERROR:
            if rcode == dns.rcode.NXDOMAIN:

                trace += '\n[NS records] {sub} does not exist.'.format(sub=sub)
            else:

                trace += '\n[NS records] Error {err}'.format(err=dns.rcode.to_text(rcode))

            break

        if len(response.authority) > 0:
            rrsets = response.authority
        elif len(response.additional) > 0:
            rrsets = [response.additional]
        else:
            rrsets = response.answer

            # Handle all RRsets, not just the first one
        for rrset in rrsets:
            for rr in rrset:
                if rr.rdtype == dns.rdatatype.SOA:
                    try:

                        trace += '\n[NS records] Same server is authoritative for {sub}'.format(sub=sub)
                    except KeyError:
                        # Here, for '1337x.unblocked.team' it returns:
                        # "unblocked.team. 300 IN SOA ns1.koaladns.com. admin.unblocked.team. 2021070507 86400 10800 604800 300"
                        return (rrset.to_text().split(' ')[4::11], trace)

                elif rr.rdtype == dns.rdatatype.A:
                    try:
                        ns = rr.items[0].address

                        trace += '\n[NS records] Glue record for {sub}: {gl}'.format(sub=rr.name, gl=ns)
                    except KeyError:
                        # {<DNS IN A rdata: 103.224.212.63>: None}
                        # Glue Record with no data

                        trace += '\n[NS records] Glue record for {sub} contains None data !'.format(sub=rr.name)
                        # [<DNS ns18.above.com. IN A RRset: [<103.224.212.63>]>, <DNS ns17.above.com. IN A RRset: [<103.224.182.63>]>]

                        # ['ns17.above.com.','3600','IN','A','103.224.182.63','ns18.above.com.','3600','IN','A','103.224.212.63']
                        # And you extract correct ns by jumping arround the list
                        return (' '.join([e.to_text() for e in rrset]).split(' ')[0::5], trace)

                elif rr.rdtype == dns.rdatatype.NS:
                    authority = rr.target

                    try:
                        if sys.version_info.major == 3:
                            ns = resolver.resolve(authority).rrset[0].to_text()
                        else:
                            ns = resolver.query(authority).rrset[0].to_text()

                        trace += '\n[NS records] {ns} ({ns_ip}) is authoritative for {sub}; ttl {ttl}'.format(ns=authority, ns_ip=ns, sub=sub, ttl=rrset.ttl)
                    except dns.resolver.NoAnswer:

                        trace += '\n[NS records] Got no answer querying {ns}'.format(ns=authority)
                        continue

                    except dns.resolver.NoNameservers:

                        trace += '\n[NS records]  All nameservers failed to answer the query {ns}. Retrying with another nameserver.'.format(ns=authority)
                        return query_authoritative_ns(domain)

                    except dns.resolver.NXDOMAIN:
                        # 1337x.full-hd-torrent.net

                        trace += '\n[NS records] {ns} does not exist'.format(ns=authority)
                        continue

                    except (dns.resolver.Timeout, dns.exception.Timeout):

                        trace += '\n[NS records] [{d}] Timeout while querying {ns}, retrying with another nameserver.'.format(d=domain, ns=authority)
                        return query_authoritative_ns(domain)

                    result = rrset.to_text()

                else:
                    # IPv6 glue records etc
                    pass

    if result is not None:
        # Here, rrset can look like:
        # <DNS fp5u7c.top. IN NS RRset: [<justin.ns.cloudflare.com.>, <dora.ns.cloudflare.com.>]>
        # <DNS bypassed.works.prx2.unblocksites.co. IN NS RRset: [<ns2.parklogic.com.>, <ns1.parklogic.com.>]>
        return (re.split(r'NS |\n', result)[1::2], trace)

    return ([], trace)

domain = 'freespt.com'
ns_querying = query_authoritative_ns(domain)
trace = '### Results for ns query_authoritative_ns("{d}"): {ns}\n{trace}'.format(
    d=domain,
    ns=ns_querying[0],
    trace=ns_querying[1]
    )

print(trace)

This produces

### Results for ns query_authoritative_ns("freespt.com"): ['ns1.bodis.com.', 'ns2.bodis.com.']
### Querying authoritative ns for freespt.com:
[NS records] [freespt.com] Used nameserver for DNS NS query is Quad9 (149.112.112.112)
[NS records] Looking up com on 149.112.112.112
[NS records] m.gtld-servers.net. (192.55.83.30) is authoritative for com; ttl 41016
[NS records] a.gtld-servers.net. (192.5.6.30) is authoritative for com; ttl 41016
[NS records] b.gtld-servers.net. (192.33.14.30) is authoritative for com; ttl 41016
[NS records] j.gtld-servers.net. (192.48.79.30) is authoritative for com; ttl 41016
[NS records] i.gtld-servers.net. (192.43.172.30) is authoritative for com; ttl 41016
[NS records] c.gtld-servers.net. (192.26.92.30) is authoritative for com; ttl 41016
[NS records] f.gtld-servers.net. (192.35.51.30) is authoritative for com; ttl 41016
[NS records] l.gtld-servers.net. (192.41.162.30) is authoritative for com; ttl 41016
[NS records] g.gtld-servers.net. (192.42.93.30) is authoritative for com; ttl 41016
[NS records] h.gtld-servers.net. (192.54.112.30) is authoritative for com; ttl 41016
[NS records] d.gtld-servers.net. (192.31.80.30) is authoritative for com; ttl 41016
[NS records] e.gtld-servers.net. (192.12.94.30) is authoritative for com; ttl 41016
[NS records] k.gtld-servers.net. (192.52.178.30) is authoritative for com; ttl 41016
[NS records] Looking up freespt.com on 192.52.178.30
[NS records] ns1.bodis.com. (199.59.242.141) is authoritative for freespt.com; ttl 172800
[NS records] ns2.bodis.com. (199.59.242.142) is authoritative for freespt.com; ttl 172800
n3rada
  • 11
  • 2