/usr/lib/python2.7/dist-packages/dnsq.py is in python-dnsq 1.1.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | '''
High-level DNS library built on top of dnspython.
Only two functions matter here:
- query_dns() : runs an arbitrary DNS query
- mx_hosts_for() : returns a list of MX hosts for a given domain
'''
import socket
import time
import dns
import dns.exception
import dns.resolver
import dns.reversename
import logging
from collections import deque
from itertools import groupby
from random import shuffle
from expiringdict import ExpiringDict
log = logging.getLogger(__name__)
# DNS resolver/cache for querying MX records
DNS_CACHE_LIFE_SECONDS=240.0
DNS_TIMEOUT_SECONDS=3.0 # timeout per DNS server
DNS_LIFETIME_TIMEOUT_SECONDS=5.2 # total timeout per DNS request
PTR_CACHE_LEN=512
# This cache is used to store PTR records for IP addresses
ptr_cache = ExpiringDict(max_len=PTR_CACHE_LEN,
max_age_seconds=DNS_CACHE_LIFE_SECONDS)
def query_dns(hostname, record_type, ns_server=None):
"""
Runs simple DNS queries, like:
>>> query_dns('mailgun.net', 'txt')
['v=spf1 include:_spf.mailgun.org ~all']
"""
try:
# if nameserver was specified, convert it into IP:
if ns_server:
ips = query_dns(ns_server, 'A')
if ips:
ns_server = ips[0]
records = exec_query(hostname, record_type, ns_server)
if record_type.lower() == 'txt':
return [record.to_text().strip("\"").replace('" "', '') for record in records]
else:
return [record.to_text() for record in records]
# no entry?
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers) as e:
return []
def mx_hosts_for(hostname):
"""
Returns a list of MX hostnames for a given domain name, sorted by their priority + randomization
Note that if no MX records are found it falls back to default
>>> mx_hosts_for('gmail.com')
['alt1.gmail-smtp-in.l.google.com', 'alt2.gmail-smtp-in.l.google.com']
Raises ecxeptions for network errors.
"""
retval = []
try:
answers = sorted(exec_query(hostname, 'MX'))
for mx_pref, grouper in groupby(answers, lambda entry: entry.preference):
group = [entry.exchange.to_text() for entry in grouper]
shuffle(group)
retval += group
# timeout, raise an exception - let them retry
except dns.exception.Timeout:
raise Exception("DNS failure for " + str(hostname))
# no MX record:
except dns.resolver.NoAnswer as err:
retval = [hostname]
# invalid domain
except dns.resolver.NXDOMAIN as err:
retval = []
# empty label (ex: domain..com)
except dns.name.EmptyLabel:
retval = []
# strip ending . and filter None
retval = [h.strip('.') for h in retval]
return filter(lambda x: x, retval)
def ptr_record_for(ipaddress):
'''
Performs reverse DNS lookup on a given IP address.
This is a replacement for socket.gethostbyaddr(), but with the following
differences:
- Returns None instead of throwing exceptions
- It is fast: it will not block for 5+ seconds for IPs without PTRs
- It is caching: it will be instant nearly all the time
>>> ptr_record_for('127.0.0.1')
"localhost"
>>> ptr_record_for('74.125.224.123')
"nuq04s08-in-f27.1e100.net"
>>> ptr_record_for('74.125.224.1')
None
'''
if ipaddress == '127.0.0.1':
return 'localhost'
MISSING = "unknown"
retval = None
# see if we have it cached:
cached_value = ptr_cache.get(ipaddress)
if cached_value:
return cached_value if cached_value != MISSING else None
try:
# get the in_addr.arpa name, like 142.224.125.74.in-addr.arpa.
inaddr_arpa_name = dns.reversename.from_address(ipaddress).to_text()
# now use ARPA name to query for PTR:
hosts = query_dns(inaddr_arpa_name, "PTR")
if hosts:
retval = hosts[0].strip('.')
ptr_cache[ipaddress] = retval
# success: found the PTR record:
return retval
except:
pass
# no suitable PTR:
ptr_cache[ipaddress] = MISSING
return None
def spf_record_for(hostname, bypass_cache=True):
"""Retrieves SPF record for a given hostname.
According to the standard, domain must not have multiple SPF records, so
if it's the case then an empty string is returned.
"""
try:
primary_ns = None
if bypass_cache:
primary_ns = get_primary_nameserver(hostname)
txt_records = query_dns(hostname, 'txt', primary_ns)
spf_records = [r for r in txt_records if r.strip().startswith('v=spf')]
if len(spf_records) == 1:
return spf_records[0]
except Exception as e:
log.exception(e)
return ''
def exec_query(hostname, record_type, ns_server=None):
"""Execute a DNS query against a given name source.
ns_server must be an IP address!!!
"""
try:
# if nameserver specified then try it first
if ns_server:
resolver = get_resolver()
resolver.nameservers = [ns_server]
try:
return resolver.query(hostname, record_type, tcp=True)
except dns.exception.Timeout:
pass
# if it's not specified or timed out then use default nameserver
return get_resolver().query(hostname, record_type, tcp=True)
# in case of timeouts and socket errors return []
except dns.exception.Timeout:
return []
except socket.error:
return []
def get_resolver():
"""Helper: return default DNS resolver object.
"""
resolver = dns.resolver.Resolver()
resolver.timeout = DNS_TIMEOUT_SECONDS
resolver.lifetime = DNS_LIFETIME_TIMEOUT_SECONDS
return resolver
def get_primary_nameserver(hostname):
"""Query DNS for the primary nameserver (SOA) for the given hostname.
"""
dq = deque(hostname.split('.'))
while len(dq) > 1:
soa = query_dns('.'.join(dq), 'SOA')
if soa:
return soa[0].split(" ")[0].strip(".")
dq.popleft()
|