in functions/source/populate_NLB_TG_with_ALB/dns/resolver.py [0:0]
def query(self, qname, rdtype=dns.rdatatype.A, rdclass=dns.rdataclass.IN,
tcp=False, source=None, raise_on_no_answer=True, source_port=0):
"""Query nameservers to find the answer to the question.
The I{qname}, I{rdtype}, and I{rdclass} parameters may be objects
of the appropriate type, or strings that can be converted into objects
of the appropriate type. E.g. For I{rdtype} the integer 2 and the
the string 'NS' both mean to query for records with DNS rdata type NS.
@param qname: the query name
@type qname: dns.name.Name object or string
@param rdtype: the query type
@type rdtype: int or string
@param rdclass: the query class
@type rdclass: int or string
@param tcp: use TCP to make the query (default is False).
@type tcp: bool
@param source: bind to this IP address (defaults to machine default
IP).
@type source: IP address in dotted quad notation
@param raise_on_no_answer: raise NoAnswer if there's no answer
(defaults is True).
@type raise_on_no_answer: bool
@param source_port: The port from which to send the message.
The default is 0.
@type source_port: int
@rtype: dns.resolver.Answer instance
@raises Timeout: no answers could be found in the specified lifetime
@raises NXDOMAIN: the query name does not exist
@raises YXDOMAIN: the query name is too long after DNAME substitution
@raises NoAnswer: the response did not contain an answer and
raise_on_no_answer is True.
@raises NoNameservers: no non-broken nameservers are available to
answer the question."""
if isinstance(qname, string_types):
qname = dns.name.from_text(qname, None)
if isinstance(rdtype, string_types):
rdtype = dns.rdatatype.from_text(rdtype)
if dns.rdatatype.is_metatype(rdtype):
raise NoMetaqueries
if isinstance(rdclass, string_types):
rdclass = dns.rdataclass.from_text(rdclass)
if dns.rdataclass.is_metaclass(rdclass):
raise NoMetaqueries
qnames_to_try = []
if qname.is_absolute():
qnames_to_try.append(qname)
else:
if len(qname) > 1:
qnames_to_try.append(qname.concatenate(dns.name.root))
if self.search:
for suffix in self.search:
qnames_to_try.append(qname.concatenate(suffix))
else:
qnames_to_try.append(qname.concatenate(self.domain))
all_nxdomain = True
nxdomain_responses = {}
start = time.time()
_qname = None # make pylint happy
for _qname in qnames_to_try:
if self.cache:
answer = self.cache.get((_qname, rdtype, rdclass))
if answer is not None:
if answer.rrset is None and raise_on_no_answer:
raise NoAnswer(response=answer.response)
else:
return answer
request = dns.message.make_query(_qname, rdtype, rdclass)
if self.keyname is not None:
request.use_tsig(self.keyring, self.keyname,
algorithm=self.keyalgorithm)
request.use_edns(self.edns, self.ednsflags, self.payload)
if self.flags is not None:
request.flags = self.flags
response = None
#
# make a copy of the servers list so we can alter it later.
#
nameservers = self.nameservers[:]
errors = []
if self.rotate:
random.shuffle(nameservers)
backoff = 0.10
while response is None:
if len(nameservers) == 0:
raise NoNameservers(request=request, errors=errors)
for nameserver in nameservers[:]:
timeout = self._compute_timeout(start)
port = self.nameserver_ports.get(nameserver, self.port)
try:
tcp_attempt = tcp
if tcp:
response = dns.query.tcp(request, nameserver,
timeout, port,
source=source,
source_port=source_port)
else:
response = dns.query.udp(request, nameserver,
timeout, port,
source=source,
source_port=source_port)
if response.flags & dns.flags.TC:
# Response truncated; retry with TCP.
tcp_attempt = True
timeout = self._compute_timeout(start)
response = \
dns.query.tcp(request, nameserver,
timeout, port,
source=source,
source_port=source_port)
except (socket.error, dns.exception.Timeout) as ex:
#
# Communication failure or timeout. Go to the
# next server
#
errors.append((nameserver, tcp_attempt, port, ex,
response))
response = None
continue
except dns.query.UnexpectedSource as ex:
#
# Who knows? Keep going.
#
errors.append((nameserver, tcp_attempt, port, ex,
response))
response = None
continue
except dns.exception.FormError as ex:
#
# We don't understand what this server is
# saying. Take it out of the mix and
# continue.
#
nameservers.remove(nameserver)
errors.append((nameserver, tcp_attempt, port, ex,
response))
response = None
continue
except EOFError as ex:
#
# We're using TCP and they hung up on us.
# Probably they don't support TCP (though
# they're supposed to!). Take it out of the
# mix and continue.
#
nameservers.remove(nameserver)
errors.append((nameserver, tcp_attempt, port, ex,
response))
response = None
continue
rcode = response.rcode()
if rcode == dns.rcode.YXDOMAIN:
ex = YXDOMAIN()
errors.append((nameserver, tcp_attempt, port, ex,
response))
raise ex
if rcode == dns.rcode.NOERROR or \
rcode == dns.rcode.NXDOMAIN:
break
#
# We got a response, but we're not happy with the
# rcode in it. Remove the server from the mix if
# the rcode isn't SERVFAIL.
#
if rcode != dns.rcode.SERVFAIL or not self.retry_servfail:
nameservers.remove(nameserver)
errors.append((nameserver, tcp_attempt, port,
dns.rcode.to_text(rcode), response))
response = None
if response is not None:
break
#
# All nameservers failed!
#
if len(nameservers) > 0:
#
# But we still have servers to try. Sleep a bit
# so we don't pound them!
#
timeout = self._compute_timeout(start)
sleep_time = min(timeout, backoff)
backoff *= 2
time.sleep(sleep_time)
if response.rcode() == dns.rcode.NXDOMAIN:
nxdomain_responses[_qname] = response
continue
all_nxdomain = False
break
if all_nxdomain:
raise NXDOMAIN(qnames=qnames_to_try, responses=nxdomain_responses)
answer = Answer(_qname, rdtype, rdclass, response,
raise_on_no_answer)
if self.cache:
self.cache.put((_qname, rdtype, rdclass), answer)
return answer