Fix DoH3 queries sent to a literal address. (#1067)

* Fix DoH3 queries sent to a literal address.

* Add test for async.

diff --git a/dns/asyncquery.py b/dns/asyncquery.py
index e5ebad41c9a282a5208b6629a85311696e108d63..e3003b1f5296a9759777e11ac8b890e06909925f 100644 (file)
--- a/dns/asyncquery.py
+++ b/dns/asyncquery.py
@@ -565,8 +565,11 @@ async def https(
             resolver = _maybe_get_resolver(resolver)
             if parsed.hostname is None:
                 raise ValueError("no hostname in URL")
-            answers = await resolver.resolve_name(parsed.hostname, family)
-            bootstrap_address = random.choice(list(answers.addresses()))
+            if dns.inet.is_address(parsed.hostname):
+                bootstrap_address = parsed.hostname
+            else:
+                answers = await resolver.resolve_name(parsed.hostname, family)
+                bootstrap_address = random.choice(list(answers.addresses()))
             if parsed.port is not None:
                 port = parsed.port
         return await _http3(

diff --git a/dns/query.py b/dns/query.py
index 39eee0ee0f97a94d35c84d80ab80da98eb412eed..bfd6908c392f60bd559e8741bd4fc378a015d731 100644 (file)
--- a/dns/query.py
+++ b/dns/query.py
@@ -439,8 +439,11 @@ def https(
             resolver = _maybe_get_resolver(resolver)
             if parsed.hostname is None:
                 raise ValueError("no hostname in URL")
-            answers = resolver.resolve_name(parsed.hostname, family)
-            bootstrap_address = random.choice(list(answers.addresses()))
+            if dns.inet.is_address(parsed.hostname):
+                bootstrap_address = parsed.hostname
+            else:
+                answers = resolver.resolve_name(parsed.hostname, family)
+                bootstrap_address = random.choice(list(answers.addresses()))
             if parsed.port is not None:
                 port = parsed.port
         return _http3(

diff --git a/tests/test_async.py b/tests/test_async.py
index 40508b0b764614952cb8ea688bd7a348f919d79a..e1cb8610f46c2a9c1fb33b5d385184be17e4c11b 100644 (file)
--- a/tests/test_async.py
+++ b/tests/test_async.py
@@ -593,6 +593,22 @@ class AsyncTests(unittest.TestCase):
 
         self.async_run(run)
 
+    @unittest.skipIf(not dns.quic.have_quic, "aioquic not available")
+    def TestDoH3QueryIP(self):
+        async def run():
+            nameserver_ip = '8.8.8.8'
+            q = dns.message.make_query("example.com.", dns.rdatatype.A)
+            r = dns.asyncquery.https(
+                q,
+                nameserver_ip,
+                post=False,
+                timeout=4,
+                h3=True,
+            )
+            self.assertTrue(q.is_response(r))
+
+        self.async_run(run)
+
     @unittest.skipIf(not dns.query._have_httpx, "httpx not available")
     def testResolverDOH(self):
         async def run():

diff --git a/tests/test_doh.py b/tests/test_doh.py
index 8912dd6cf135f9108091cc44c6fe4104a8cf83a5..692b2d670d890bd1aa16c24998244726c88673ff 100644 (file)
--- a/tests/test_doh.py
+++ b/tests/test_doh.py
@@ -220,6 +220,33 @@ class DNSOverHTTP3TestCase(unittest.TestCase):
         )
         self.assertTrue(q.is_response(r))
 
+    def test_build_url_from_ip(self):
+        self.assertTrue(resolver_v4_addresses or resolver_v6_addresses)
+        if resolver_v4_addresses:
+            nameserver_ip = random.choice(resolver_v4_addresses)
+            q = dns.message.make_query("example.com.", dns.rdatatype.A)
+            # For some reason Google's DNS over HTTPS fails when you POST to
+            # https://8.8.8.8/dns-query
+            # So we're just going to do GET requests here
+            r = dns.query.https(
+                q,
+                nameserver_ip,
+                post=False,
+                timeout=4,
+                h3=True,
+            )
+            self.assertTrue(q.is_response(r))
+        if resolver_v6_addresses:
+            nameserver_ip = random.choice(resolver_v6_addresses)
+            q = dns.message.make_query("example.com.", dns.rdatatype.A)
+            r = dns.query.https(
+                q,
+                nameserver_ip,
+                post=False,
+                timeout=4,
+                h3=True,
+            )
+            self.assertTrue(q.is_response(r))
 
 if __name__ == "__main__":
     unittest.main()