name = name.parent()
except dns.name.NoParent: # pragma: no cover
raise NoRootSOA
+
+
+async def make_resolver_at(
+ where: Union[dns.name.Name, str],
+ port: int = 53,
+ family: int = socket.AF_UNSPEC,
+ resolver: Optional[Resolver] = None,
+) -> Resolver:
+ """Make a stub resolver using the specified destination as the full resolver.
+
+ *where*, a ``dns.name.Name`` or ``str`` the domain name or IP address of the
+ full resolver.
+
+ *port*, an ``int``, the port to use. If not specified, the default is 53.
+
+ *family*, an ``int``, the address family to use. This parameter is used if
+ *where* is not an address. The default is ``socket.AF_UNSPEC`` in which case
+ the first address returned by ``resolve_name()`` will be used, otherwise the
+ first address of the specified family will be used.
+
+ *resolver*, a ``dns.asyncresolver.Resolver`` or ``None``, the resolver to use for
+ resolution of hostnames. If not specified, the default resolver will be used.
+
+ Returns a ``dns.resolver.Resolver`` or raises an exception.
+ """
+ if resolver is None:
+ resolver = get_default_resolver()
+ nameservers = []
+ if isinstance(where, str) and dns.inet.is_address(where):
+ nameservers.append(dns.nameserver.Do53Nameserver(where, port))
+ else:
+ answers = await resolver.resolve_name(where, family)
+ for address in answers.addresses():
+ nameservers.append(dns.nameserver.Do53Nameserver(address, port))
+ res = dns.asyncresolver.Resolver(configure=False)
+ res.nameservers = nameservers
+ return res
+
+
+async def resolve_at(
+ where: Union[dns.name.Name, str],
+ qname: Union[dns.name.Name, str],
+ rdtype: Union[dns.rdatatype.RdataType, str] = dns.rdatatype.A,
+ rdclass: Union[dns.rdataclass.RdataClass, str] = dns.rdataclass.IN,
+ tcp: bool = False,
+ source: Optional[str] = None,
+ raise_on_no_answer: bool = True,
+ source_port: int = 0,
+ lifetime: Optional[float] = None,
+ search: Optional[bool] = None,
+ backend: Optional[dns.asyncbackend.Backend] = None,
+ port: int = 53,
+ family: int = socket.AF_UNSPEC,
+ resolver: Optional[Resolver] = None,
+) -> dns.resolver.Answer:
+ """Query nameservers to find the answer to the question.
+
+ This is a convenience function that calls ``dns.asyncresolver.make_resolver_at()``
+ to make a resolver, and then uses it to resolve the query.
+
+ See ``dns.asyncresolver.Resolver.resolve`` for more information on the resolution
+ parameters, and ``dns.asyncresolver.make_resolver_at`` for information about the
+ resolver parameters *where*, *port*, *family*, and *resolver*.
+
+ If making more than one query, it is more efficient to call
+ ``dns.asyncresolver.make_resolver_at()`` and then use that resolver for the queries
+ instead of calling ``resolve_at()`` multiple times.
+ """
+ res = await make_resolver_at(where, port, family, resolver)
+ return await res.resolve(
+ qname,
+ rdtype,
+ rdclass,
+ tcp,
+ source,
+ raise_on_no_answer,
+ source_port,
+ lifetime,
+ search,
+ backend,
+ )
raise NoRootSOA
+def make_resolver_at(
+ where: Union[dns.name.Name, str],
+ port: int = 53,
+ family: int = socket.AF_UNSPEC,
+ resolver: Optional[Resolver] = None,
+) -> Resolver:
+ """Make a stub resolver using the specified destination as the full resolver.
+
+ *where*, a ``dns.name.Name`` or ``str`` the domain name or IP address of the
+ full resolver.
+
+ *port*, an ``int``, the port to use. If not specified, the default is 53.
+
+ *family*, an ``int``, the address family to use. This parameter is used if
+ *where* is not an address. The default is ``socket.AF_UNSPEC`` in which case
+ the first address returned by ``resolve_name()`` will be used, otherwise the
+ first address of the specified family will be used.
+
+ *resolver*, a ``dns.resolver.Resolver`` or ``None``, the resolver to use for
+ resolution of hostnames. If not specified, the default resolver will be used.
+
+ Returns a ``dns.resolver.Resolver`` or raises an exception.
+ """
+ if resolver is None:
+ resolver = get_default_resolver()
+ nameservers = []
+ if isinstance(where, str) and dns.inet.is_address(where):
+ nameservers.append(dns.nameserver.Do53Nameserver(where, port))
+ else:
+ for address in resolver.resolve_name(where, family).addresses():
+ nameservers.append(dns.nameserver.Do53Nameserver(address, port))
+ res = dns.resolver.Resolver(configure=False)
+ res.nameservers = nameservers
+ return res
+
+
+def resolve_at(
+ where: Union[dns.name.Name, str],
+ qname: Union[dns.name.Name, str],
+ rdtype: Union[dns.rdatatype.RdataType, str] = dns.rdatatype.A,
+ rdclass: Union[dns.rdataclass.RdataClass, str] = dns.rdataclass.IN,
+ tcp: bool = False,
+ source: Optional[str] = None,
+ raise_on_no_answer: bool = True,
+ source_port: int = 0,
+ lifetime: Optional[float] = None,
+ search: Optional[bool] = None,
+ port: int = 53,
+ family: int = socket.AF_UNSPEC,
+ resolver: Optional[Resolver] = None,
+) -> Answer:
+ """Query nameservers to find the answer to the question.
+
+ This is a convenience function that calls ``dns.resolver.make_resolver_at()`` to
+ make a resolver, and then uses it to resolve the query.
+
+ See ``dns.resolver.Resolver.resolve`` for more information on the resolution
+ parameters, and ``dns.resolver.make_resolver_at`` for information about the resolver
+ parameters *where*, *port*, *family*, and *resolver*.
+
+ If making more than one query, it is more efficient to call
+ ``dns.resolver.make_resolver_at()`` and then use that resolver for the queries
+ instead of calling ``resolve_at()`` multiple times.
+ """
+ return make_resolver_at(where, port, family, resolver).resolve(
+ qname,
+ rdtype,
+ rdclass,
+ tcp,
+ source,
+ raise_on_no_answer,
+ source_port,
+ lifetime,
+ search,
+ )
+
+
#
# Support for overriding the system resolver for all python code in the
# running process.
.. autofunction:: dns.asyncresolver.canonical_name
.. autofunction:: dns.asyncresolver.try_ddr
.. autofunction:: dns.asyncresolver.zone_for_name
+.. autofunction:: dns.asyncresolver.make_resolver_at
+.. autofunction:: dns.asyncresolver.resolve_at
.. autodata:: dns.asyncresolver.default_resolver
.. autofunction:: dns.asyncresolver.get_default_resolver
.. autofunction:: dns.asyncresolver.reset_default_resolver
.. autofunction:: dns.resolver.try_ddr
.. autofunction:: dns.resolver.zone_for_name
.. autofunction:: dns.resolver.query
+.. autofunction:: dns.resolver.make_resolver_at
+.. autofunction:: dns.resolver.resolve_at
.. autodata:: dns.resolver.default_resolver
.. autofunction:: dns.resolver.get_default_resolver
.. autofunction:: dns.resolver.reset_default_resolver
DNS-over-QUIC. This feature is currently experimental as the standard is still in
draft stage.
+* The resolver and async resolver now have the ``make_resolver_at()`` and
+ ``resolve_at()`` functions, as a convenience for making queries to specific
+ recursive servers.
+
* Curio support has been removed.
2.3.0
print(a.response)
zn = await dns.asyncresolver.zone_for_name(host)
print(zn)
+ answer = await dns.asyncresolver.resolve_at("8.8.8.8", "amazon.com", "NS")
+ print("The amazon.com nameservers are:")
+ for rr in answer:
+ print(rr.target)
if __name__ == "__main__":
print("")
print("")
-# A higher-level way
+# A higher-level way:
import dns.resolver
-resolver = dns.resolver.Resolver(configure=False)
-resolver.nameservers = ["8.8.8.8"]
-answer = resolver.resolve("amazon.com", "NS")
+answer = dns.resolver.resolve_at("8.8.8.8", "amazon.com", "NS")
print("The nameservers are:")
for rr in answer:
print(rr.target)
+print("")
+print("")
+
+# If you're going to make a bunch of queries to the server, make the resolver once
+# and then use it multiple times:
+
+res = dns.resolver.make_resolver_at("dns.google")
+answer = res.resolve("amazon.com", "NS")
+print("The amazon.com nameservers are:")
+for rr in answer:
+ print(rr.target)
+answer = res.resolve("google.com", "NS")
+print("The google.com nameservers are:")
+for rr in answer:
+ print(rr.target)
+print("")
+print("")
# Sending a query with the all flags set to 0. This is the easiest way
# to make a query with the RD flag off.
self.async_run(run)
+ @unittest.skipIf(not tests.util.have_ipv4(), "IPv4 not reachable")
+ def testResolveAtAddress(self):
+ async def run():
+ answer = await dns.asyncresolver.resolve_at("8.8.8.8", "dns.google.", "A")
+ seen = set([rdata.address for rdata in answer])
+ self.assertIn("8.8.8.8", seen)
+ self.assertIn("8.8.4.4", seen)
+
+ self.async_run(run)
+
+ @unittest.skipIf(not tests.util.have_ipv4(), "IPv4 not reachable")
+ def testResolveAtName(self):
+ async def run():
+ answer = await dns.asyncresolver.resolve_at(
+ "dns.google", "dns.google.", "A", family=socket.AF_INET
+ )
+ seen = set([rdata.address for rdata in answer])
+ self.assertIn("8.8.8.8", seen)
+ self.assertIn("8.8.4.4", seen)
+
+ self.async_run(run)
+
def testSleep(self):
async def run():
before = time.time()
self.assertIn("94.140.14.14", seen)
self.assertIn("94.140.15.15", seen)
+ @unittest.skipIf(not tests.util.have_ipv4(), "IPv4 not reachable")
+ def testResolveAtAddress(self):
+ answer = dns.resolver.resolve_at("8.8.8.8", "dns.google.", "A")
+ seen = set([rdata.address for rdata in answer])
+ self.assertIn("8.8.8.8", seen)
+ self.assertIn("8.8.4.4", seen)
+
+ @unittest.skipIf(not tests.util.have_ipv4(), "IPv4 not reachable")
+ def testResolveAtName(self):
+ answer = dns.resolver.resolve_at(
+ "dns.google", "dns.google.", "A", family=socket.AF_INET
+ )
+ seen = set([rdata.address for rdata in answer])
+ self.assertIn("8.8.8.8", seen)
+ self.assertIn("8.8.4.4", seen)
+
def testCanonicalNameNoCNAME(self):
cname = dns.name.from_text("www.google.com")
self.assertEqual(dns.resolver.canonical_name("www.google.com"), cname)
projects / thirdparty / dnspython.git / commitdiff
