From: Brian Wellington <bwelling@xbill.org>
Date: Wed, 20 Mar 2024 19:25:13 +0000 (-0700)
Subject: Fix DoH3 queries sent to a literal address. (#1067)
X-Git-Tag: v2.7.0rc1~62
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=23a6cd2a6d8b1d9612edd11261870acd702c3136;p=thirdparty%2Fdnspython.git

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 e5ebad41..e3003b1f 100644
--- 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 39eee0ee..bfd6908c 100644
--- 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 40508b0b..e1cb8610 100644
--- 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 8912dd6c..692b2d67 100644
--- 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()