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

* Fix DoH3 queries sent to a literal address. * Add test for async.
rthalley · Mar 20, 2024 · 23a6cd2 · 23a6cd2
1 parent 6514e04
commit 23a6cd2
Show file tree

Hide file tree

Showing 4 changed files with 53 additions and 4 deletions.
diff --git 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
@@ -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
@@ -593,6 +593,22 @@ async def run():
 
         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
@@ -220,6 +220,33 @@ def testDoH3PostRequest(self):
         )
         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()