File CVE-2023-29483.patch of Package python-dnspython

Overview Repositories Revisions Requests Users Attributes Meta

File CVE-2023-29483.patch of Package python-dnspython (Revision 7cfcc7fe3abba23fc2110b8c6294f9e8)

Currently displaying revision 7cfcc7fe3abba23fc2110b8c6294f9e8 , Show latest

From 093c593624bcf55766c2a952c207e0b92920214e Mon Sep 17 00:00:00 2001
From: Bob Halley <halley@dnspython.org>
Date: Fri, 9 Feb 2024 10:36:08 -0800
Subject: [PATCH] Address DoS via the Tudoor mechanism (CVE-2023-29483)

---
 dns/asyncquery.py |  45 +++++++++++++------
 dns/nameserver.py |   2 +
 dns/query.py      | 110 +++++++++++++++++++++++++++++-----------------
 3 files changed, 103 insertions(+), 54 deletions(-)

Index: dnspython-1.12.0/dns/query.py
===================================================================
--- dnspython-1.12.0.orig/dns/query.py
+++ dnspython-1.12.0/dns/query.py
@@ -155,18 +155,38 @@ def _wait_for_writable(s, expiration):
     _wait_for(s, False, True, True, expiration)
 
 
+def _matches_destination(af, from_address, destination, ignore_unexpected):
+    # Check that from_address is appropriate for a response to a query
+    # sent to destination.
+    if not destination:
+        return True
+    if _addresses_equal(af, from_address, destination) or (
+        dns.inet.is_multicast(destination[0]) and from_address[1:] == destination[1:]
+    ):
+        return True
+    elif ignore_unexpected:
+        return False
+    raise UnexpectedSource(
+        "got a response from %s instead of %s" % (from_address, destination)
+    )
+
+
 def _addresses_equal(af, a1, a2):
     # Convert the first value of the tuple, which is a textual format
     # address into binary form, so that we are not confused by different
     # textual representations of the same address
-    n1 = dns.inet.inet_pton(af, a1[0])
-    n2 = dns.inet.inet_pton(af, a2[0])
+    try:
+        n1 = dns.inet.inet_pton(af, a1[0])
+        n2 = dns.inet.inet_pton(af, a2[0])
+    except dns.exception.SyntaxError:
+        return False
     return n1 == n2 and a1[1:] == a2[1:]
 
 
 def _destination_and_source(af, where, port, source, source_port):
     # Apply defaults and compute destination and source tuples
     # suitable for use in connect(), sendto(), or bind().
+    destination = None
     if af is None:
         try:
             af = dns.inet.af_for_address(where)
@@ -187,6 +207,33 @@ def _destination_and_source(af, where, p
     return (af, destination, source)
 
 
+def _udp_recv(sock, max_size, expiration):
+    """Reads a datagram from the socket.
+    A Timeout exception will be raised if the operation is not completed
+    by the expiration time.
+    """
+    while True:
+        try:
+            return sock.recvfrom(max_size)
+        except BlockingIOError:
+            _wait_for_readable(sock, expiration)
+
+
+def _udp_send(sock, data, destination, expiration):
+    """Sends the specified datagram to destination over the socket.
+    A Timeout exception will be raised if the operation is not completed
+    by the expiration time.
+    """
+    while True:
+        try:
+            if destination:
+                return sock.sendto(data, destination)
+            else:
+                return sock.send(data)
+        except BlockingIOError:  # pragma: no cover
+            _wait_for_writable(sock, expiration)
+
+
 def send_udp(sock, what, destination, expiration=None):
     """Send a DNS message to the specified UDP socket.
 
@@ -203,15 +250,15 @@ def send_udp(sock, what, destination, ex
     """
     if isinstance(what, dns.message.Message):
         what = what.to_wire()
-    _wait_for_writable(sock, expiration)
     sent_time = time.time()
-    n = sock.sendto(what, destination)
+    n = _udp_send(sock, what, destination, expiration)
     return (n, sent_time)
 
 
 def receive_udp(sock, destination, expiration=None, af=None,
                 ignore_unexpected=False, one_rr_per_rrset=False,
-                keyring=None, request_mac=b''):
+                keyring=None, request_mac=b'',
+                ignore_errors=False, query=None):
     """Read a DNS message from a UDP socket.
 
     @param sock: the socket
@@ -229,6 +276,14 @@ def receive_udp(sock, destination, expir
     @type keyring: keyring dict
     @param request_mac: the MAC of the request (for TSIG)
     @type request_mac: bytes
+    @param ignore_errors: If various format errors or response
+    mismatches occur, ignore them and keep listening for a valid response.
+    The default is ``False``.
+    @type ignore_errors: bool
+    @param query: If not ``None`` and *ignore_errors* is ``True``,
+    check that the received message is a response to this query, and
+    if not keep listening for a valid response.
+    @type query: dns.message.Message or None
     @rtype: dns.message.Message object
     """
     if af is None:
@@ -238,23 +293,32 @@ def receive_udp(sock, destination, expir
             af = dns.inet.AF_INET
     wire = b''
     while 1:
-        _wait_for_readable(sock, expiration)
-        (wire, from_address) = sock.recvfrom(65535)
-        if _addresses_equal(af, from_address, destination) or \
-           (dns.inet.is_multicast(destination[0]) and
-            from_address[1:] == destination[1:]):
-            break
-        if not ignore_unexpected:
-            raise UnexpectedSource('got a response from '
-                                   '%s instead of %s' % (from_address,
-                                                         destination))
-    received_time = time.time()
-    r = dns.message.from_wire(wire, keyring=keyring, request_mac=request_mac,
-                              one_rr_per_rrset=one_rr_per_rrset)
-    return (r, received_time)
+        (wire, from_address) = _udp_recv(sock, 65535, expiration)
+        if not _matches_destination(
+             af, from_address, destination, ignore_unexpected
+         ):
+            continue
+
+        received_time = time.time()
+        try:
+            r = dns.message.from_wire(
+                wire,
+                keyring=keyring,
+                request_mac=request_mac,
+                one_rr_per_rrset=one_rr_per_rrset,
+            )
+        except Exception:
+            if ignore_errors:
+                continue
+            else:
+                raise
+        if ignore_errors and query is not None and not query.is_response(r):
+            continue
+        return (r, received_time)
 
 def udp(q, where, timeout=None, port=53, af=None, source=None, source_port=0,
-        ignore_unexpected=False, one_rr_per_rrset=False):
+        ignore_unexpected=False, one_rr_per_rrset=False,
+        ignore_errors=False, sock=None):
     """Return the response obtained after sending a query via UDP.
 
     @param q: the query
@@ -280,13 +344,20 @@ def udp(q, where, timeout=None, port=53,
     @type ignore_unexpected: bool
     @param one_rr_per_rrset: Put each RR into its own RRset
     @type one_rr_per_rrset: bool
+    @param ignore_errors: If various format errors or response
+    mismatches occur, ignore them and keep listening for a valid
+    response. The default is ``False``.
+    @type ignore_errors: bool
     @rtype: dns.message.Message object
     """
 
     wire = q.to_wire()
     (af, destination, source) = _destination_and_source(af, where, port,
                                                         source, source_port)
-    s = socket.socket(af, socket.SOCK_DGRAM, 0)
+    if sock:
+        s = sock
+    else:
+        s = socket.socket(af, socket.SOCK_DGRAM, 0)
     received_time = None
     sent_time = None
     try:
@@ -298,10 +369,13 @@ def udp(q, where, timeout=None, port=53,
         (_, sent_time) = send_udp(s, wire, destination, expiration)
         (r, received_time) = receive_udp(s, destination, expiration, af,
                                          ignore_unexpected, one_rr_per_rrset,
-                                         q.keyring, q.request_mac)
+                                         q.keyring, q.request_mac,
+                                         ignore_errors, q)
     finally:
         s.close()
-    if not q.is_response(r):
+    # We don't need to check q.is_response() if we are in ignore_errors mode
+    # as receive_udp() will have checked it.
+    if not (ignore_errors or q.is_response(r)):
         raise BadResponse
     return r
 
@@ -520,8 +594,7 @@ def xfr(where, zone, rdtype=dns.rdatatyp
     _connect(s, destination)
     l = len(wire)
     if use_udp:
-        _wait_for_writable(s, expiration)
-        s.send(wire)
+        _udp_send(s, wire, None, expiration)
     else:
         tcpmsg = struct.pack("!H", l) + wire
         _net_write(s, tcpmsg, expiration)
@@ -542,8 +615,7 @@ def xfr(where, zone, rdtype=dns.rdatatyp
         if mexpiration is None or mexpiration > expiration:
             mexpiration = expiration
         if use_udp:
-            _wait_for_readable(s, expiration)
-            (wire, from_address) = s.recvfrom(65535)
+            (wire, from_address) = _udp_recv(s, 65535, expiration)
         else:
             ldata = _net_read(s, 2, mexpiration)
             (l,) = struct.unpack("!H", ldata)
Index: dnspython-1.12.0/dns/resolver.py
===================================================================
--- dnspython-1.12.0.orig/dns/resolver.py
+++ dnspython-1.12.0/dns/resolver.py
@@ -850,7 +850,9 @@ class Resolver(object):
                             response = dns.query.udp(request, nameserver,
                                                      timeout, self.port,
                                                      source=source,
-                                                     source_port=source_port)
+                                                     source_port=source_port,
+                                                     ignore_errors=True,
+                                                     ignore_unexpected=True)
                             if response.flags & dns.flags.TC:
                                 # Response truncated; retry with TCP.
                                 timeout = self._compute_timeout(start)
Index: dnspython-1.12.0/tests/test_query.py
===================================================================
--- /dev/null
+++ dnspython-1.12.0/tests/test_query.py
@@ -0,0 +1,758 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-2017 Nominum, Inc.
+#
+# Permission to use, copy, modify, and distribute this software and its
+# documentation for any purpose with or without fee is hereby granted,
+# provided that the above copyright notice and this permission notice
+# appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
+# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
+# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
+# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+import contextlib
+import socket
+import sys
+import time
+import unittest
+
+try:
+    import ssl
+
+    have_ssl = True
+except Exception:
+    have_ssl = False
+
+import dns.exception
+import dns.flags
+import dns.inet
+import dns.message
+import dns.name
+import dns.query
+import dns.rcode
+import dns.rdataclass
+import dns.rdatatype
+import dns.tsigkeyring
+import dns.zone
+
+_nanonameserver_available = False
+
+class Server(object):
+    pass
+
+
+query_addresses = []
+
+keyring = dns.tsigkeyring.from_text({"name": b"tDz6cfXXGtNivRpQ98hr6A=="})
+
+
+@unittest.skipIf(True, "Internet not reachable")
+class QueryTests(unittest.TestCase):
+    def testQueryUDP(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A)
+            response = dns.query.udp(q, address, timeout=2)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("8.8.8.8" in seen)
+            self.assertTrue("8.8.4.4" in seen)
+
+    def testQueryUDPWithSocket(self):
+        for address in query_addresses:
+            with socket.socket(
+                dns.inet.af_for_address(address), socket.SOCK_DGRAM
+            ) as s:
+                s.setblocking(0)
+                qname = dns.name.from_text("dns.google.")
+                q = dns.message.make_query(qname, dns.rdatatype.A)
+                response = dns.query.udp(q, address, sock=s, timeout=2)
+                rrs = response.get_rrset(
+                    response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+                )
+                self.assertTrue(rrs is not None)
+                seen = set([rdata.address for rdata in rrs])
+                self.assertTrue("8.8.8.8" in seen)
+                self.assertTrue("8.8.4.4" in seen)
+
+    def testQueryTCP(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A)
+            response = dns.query.tcp(q, address, timeout=2)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("8.8.8.8" in seen)
+            self.assertTrue("8.8.4.4" in seen)
+
+    def testQueryTCPWithSocket(self):
+        for address in query_addresses:
+            with socket.socket(
+                dns.inet.af_for_address(address), socket.SOCK_STREAM
+            ) as s:
+                ll = dns.inet.low_level_address_tuple((address, 53))
+                s.settimeout(2)
+                s.connect(ll)
+                s.setblocking(0)
+                qname = dns.name.from_text("dns.google.")
+                q = dns.message.make_query(qname, dns.rdatatype.A)
+                response = dns.query.tcp(q, None, sock=s, timeout=2)
+                rrs = response.get_rrset(
+                    response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+                )
+                self.assertTrue(rrs is not None)
+                seen = set([rdata.address for rdata in rrs])
+                self.assertTrue("8.8.8.8" in seen)
+                self.assertTrue("8.8.4.4" in seen)
+
+    @unittest.skipUnless(have_ssl, "No SSL support")
+    def testQueryTLS(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A)
+            response = dns.query.tls(q, address, timeout=2)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("8.8.8.8" in seen)
+            self.assertTrue("8.8.4.4" in seen)
+
+    @unittest.skipUnless(have_ssl, "No SSL support")
+    def testQueryTLSWithContext(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A)
+            ssl_context = ssl.create_default_context()
+            ssl_context.check_hostname = False
+            response = dns.query.tls(q, address, timeout=2, ssl_context=ssl_context)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("8.8.8.8" in seen)
+            self.assertTrue("8.8.4.4" in seen)
+
+    @unittest.skipUnless(have_ssl, "No SSL support")
+    def testQueryTLSWithSocket(self):
+        for address in query_addresses:
+            with socket.socket(
+                dns.inet.af_for_address(address), socket.SOCK_STREAM
+            ) as base_s:
+                ll = dns.inet.low_level_address_tuple((address, 853))
+                base_s.settimeout(2)
+                base_s.connect(ll)
+                ctx = ssl.create_default_context()
+                ctx.minimum_version = ssl.TLSVersion.TLSv1_2
+                with ctx.wrap_socket(
+                    base_s, server_hostname="dns.google"
+                ) as s:  # lgtm[py/insecure-protocol]
+                    s.setblocking(0)
+                    qname = dns.name.from_text("dns.google.")
+                    q = dns.message.make_query(qname, dns.rdatatype.A)
+                    response = dns.query.tls(q, None, sock=s, timeout=2)
+                    rrs = response.get_rrset(
+                        response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+                    )
+                    self.assertTrue(rrs is not None)
+                    seen = set([rdata.address for rdata in rrs])
+                    self.assertTrue("8.8.8.8" in seen)
+                    self.assertTrue("8.8.4.4" in seen)
+
+    @unittest.skipUnless(have_ssl, "No SSL support")
+    def testQueryTLSwithPadding(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A, use_edns=0, pad=128)
+            response = dns.query.tls(q, address, timeout=2)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("8.8.8.8" in seen)
+            self.assertTrue("8.8.4.4" in seen)
+            # the response should have a padding option
+            self.assertIsNotNone(response.opt)
+            has_pad = False
+            for o in response.opt[0].options:
+                if o.otype == dns.edns.OptionType.PADDING:
+                    has_pad = True
+            self.assertTrue(has_pad)
+
+    def testQueryUDPFallback(self):
+        for address in query_addresses:
+            qname = dns.name.from_text(".")
+            q = dns.message.make_query(qname, dns.rdatatype.DNSKEY)
+            (_, tcp) = dns.query.udp_with_fallback(q, address, timeout=4)
+            self.assertTrue(tcp)
+
+    def testQueryUDPFallbackWithSocket(self):
+        for address in query_addresses:
+            af = dns.inet.af_for_address(address)
+            with socket.socket(af, socket.SOCK_DGRAM) as udp_s:
+                udp_s.setblocking(0)
+                with socket.socket(af, socket.SOCK_STREAM) as tcp_s:
+                    ll = dns.inet.low_level_address_tuple((address, 53))
+                    tcp_s.settimeout(2)
+                    tcp_s.connect(ll)
+                    tcp_s.setblocking(0)
+                    qname = dns.name.from_text(".")
+                    q = dns.message.make_query(qname, dns.rdatatype.DNSKEY)
+                    (_, tcp) = dns.query.udp_with_fallback(
+                        q, address, udp_sock=udp_s, tcp_sock=tcp_s, timeout=4
+                    )
+                    self.assertTrue(tcp)
+
+    def testQueryUDPFallbackNoFallback(self):
+        for address in query_addresses:
+            qname = dns.name.from_text("dns.google.")
+            q = dns.message.make_query(qname, dns.rdatatype.A)
+            (_, tcp) = dns.query.udp_with_fallback(q, address, timeout=2)
+            self.assertFalse(tcp)
+
+    def testUDPReceiveQuery(self):
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as listener:
+            listener.bind(("127.0.0.1", 0))
+            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sender:
+                sender.bind(("127.0.0.1", 0))
+                q = dns.message.make_query("dns.google", dns.rdatatype.A)
+                dns.query.send_udp(sender, q, listener.getsockname())
+                expiration = time.time() + 2
+                (q, _, addr) = dns.query.receive_udp(listener, expiration=expiration)
+                self.assertEqual(addr, sender.getsockname())
+
+
+axfr_zone = """
+$TTL 300
+@ SOA ns1 root 1 7200 900 1209600 86400
+@ NS ns1
+@ NS ns2
+ns1 A 10.0.0.1
+ns2 A 10.0.0.1
+"""
+
+
+class AXFRNanoNameserver(Server):
+    def handle(self, request):
+        self.zone = dns.zone.from_text(axfr_zone, origin=self.origin)
+        self.origin = self.zone.origin
+        items = []
+        soa = self.zone.find_rrset(dns.name.empty, dns.rdatatype.SOA)
+        response = dns.message.make_response(request.message)
+        response.flags |= dns.flags.AA
+        response.answer.append(soa)
+        items.append(response)
+        response = dns.message.make_response(request.message)
+        response.question = []
+        response.flags |= dns.flags.AA
+        for name, rdataset in self.zone.iterate_rdatasets():
+            if rdataset.rdtype == dns.rdatatype.SOA and name == dns.name.empty:
+                continue
+            rrset = dns.rrset.RRset(
+                name, rdataset.rdclass, rdataset.rdtype, rdataset.covers
+            )
+            rrset.update(rdataset)
+            response.answer.append(rrset)
+        items.append(response)
+        response = dns.message.make_response(request.message)
+        response.question = []
+        response.flags |= dns.flags.AA
+        response.answer.append(soa)
+        items.append(response)
+        return items
+
+
+ixfr_message = """id 12345
+opcode QUERY
+rcode NOERROR
+flags AA
+;QUESTION
+example. IN IXFR
+;ANSWER
+example. 300 IN SOA ns1.example. root.example. 4 7200 900 1209600 86400
+example. 300 IN SOA ns1.example. root.example. 2 7200 900 1209600 86400
+deleted.example. 300 IN A 10.0.0.1
+changed.example. 300 IN A 10.0.0.2
+example. 300 IN SOA ns1.example. root.example. 3 7200 900 1209600 86400
+changed.example. 300 IN A 10.0.0.4
+added.example. 300 IN A 10.0.0.3
+example. 300 SOA ns1.example. root.example. 3 7200 900 1209600 86400
+example. 300 IN SOA ns1.example. root.example. 4 7200 900 1209600 86400
+added2.example. 300 IN A 10.0.0.5
+example. 300 IN SOA ns1.example. root.example. 4 7200 900 1209600 86400
+"""
+
+ixfr_trailing_junk = ixfr_message + "junk.example. 300 IN A 10.0.0.6"
+
+ixfr_up_to_date_message = """id 12345
+opcode QUERY
+rcode NOERROR
+flags AA
+;QUESTION
+example. IN IXFR
+;ANSWER
+example. 300 IN SOA ns1.example. root.example. 2 7200 900 1209600 86400
+"""
+
+axfr_trailing_junk = """id 12345
+opcode QUERY
+rcode NOERROR
+flags AA
+;QUESTION
+example. IN AXFR
+;ANSWER
+example. 300 IN SOA ns1.example. root.example. 3 7200 900 1209600 86400
+added.example. 300 IN A 10.0.0.3
+added2.example. 300 IN A 10.0.0.5
+changed.example. 300 IN A 10.0.0.4
+example. 300 IN SOA ns1.example. root.example. 3 7200 900 1209600 86400
+junk.example. 300 IN A 10.0.0.6
+"""
+
+
+class IXFRNanoNameserver(Server):
+    def __init__(self, response_text):
+        super().__init__()
+        self.response_text = response_text
+
+    def handle(self, request):
+        try:
+            r = dns.message.from_text(self.response_text, one_rr_per_rrset=True)
+            r.id = request.message.id
+            return r
+        except Exception:
+            pass
+
+
+@unittest.skipIf(not _nanonameserver_available, "nanonameserver required")
+class XfrTests(unittest.TestCase):
+    def test_axfr(self):
+        expected = dns.zone.from_text(axfr_zone, origin="example")
+        with AXFRNanoNameserver(origin="example") as ns:
+            xfr = dns.query.xfr(ns.tcp_address[0], "example", port=ns.tcp_address[1])
+            zone = dns.zone.from_xfr(xfr)
+            self.assertEqual(zone, expected)
+
+    def test_axfr_tsig(self):
+        expected = dns.zone.from_text(axfr_zone, origin="example")
+        with AXFRNanoNameserver(origin="example", keyring=keyring) as ns:
+            xfr = dns.query.xfr(
+                ns.tcp_address[0],
+                "example",
+                port=ns.tcp_address[1],
+                keyring=keyring,
+                keyname="name",
+            )
+            zone = dns.zone.from_xfr(xfr)
+            self.assertEqual(zone, expected)
+
+    def test_axfr_root_tsig(self):
+        expected = dns.zone.from_text(axfr_zone, origin=".")
+        with AXFRNanoNameserver(origin=".", keyring=keyring) as ns:
+            xfr = dns.query.xfr(
+                ns.tcp_address[0],
+                ".",
+                port=ns.tcp_address[1],
+                keyring=keyring,
+                keyname="name",
+            )
+            zone = dns.zone.from_xfr(xfr)
+            self.assertEqual(zone, expected)
+
+    def test_axfr_udp(self):
+        def bad():
+            with AXFRNanoNameserver(origin="example") as ns:
+                xfr = dns.query.xfr(
+                    ns.udp_address[0], "example", port=ns.udp_address[1], use_udp=True
+                )
+                l = list(xfr)
+
+        self.assertRaises(ValueError, bad)
+
+    def test_axfr_bad_rcode(self):
+        def bad():
+            # We just use Server here as by default it will refuse.
+            with Server() as ns:
+                xfr = dns.query.xfr(
+                    ns.tcp_address[0], "example", port=ns.tcp_address[1]
+                )
+                l = list(xfr)
+
+        self.assertRaises(dns.query.TransferError, bad)
+
+    def test_axfr_trailing_junk(self):
+        # we use the IXFR server here as it returns messages
+        def bad():
+            with IXFRNanoNameserver(axfr_trailing_junk) as ns:
+                xfr = dns.query.xfr(
+                    ns.tcp_address[0],
+                    "example",
+                    dns.rdatatype.AXFR,
+                    port=ns.tcp_address[1],
+                )
+                l = list(xfr)
+
+        self.assertRaises(dns.exception.FormError, bad)
+
+    def test_ixfr_tcp(self):
+        with IXFRNanoNameserver(ixfr_message) as ns:
+            xfr = dns.query.xfr(
+                ns.tcp_address[0],
+                "example",
+                dns.rdatatype.IXFR,
+                port=ns.tcp_address[1],
+                serial=2,
+                relativize=False,
+            )
+            l = list(xfr)
+            self.assertEqual(len(l), 1)
+            expected = dns.message.from_text(ixfr_message, one_rr_per_rrset=True)
+            expected.id = l[0].id
+            self.assertEqual(l[0], expected)
+
+    def test_ixfr_udp(self):
+        with IXFRNanoNameserver(ixfr_message) as ns:
+            xfr = dns.query.xfr(
+                ns.udp_address[0],
+                "example",
+                dns.rdatatype.IXFR,
+                port=ns.udp_address[1],
+                serial=2,
+                relativize=False,
+                use_udp=True,
+            )
+            l = list(xfr)
+            self.assertEqual(len(l), 1)
+            expected = dns.message.from_text(ixfr_message, one_rr_per_rrset=True)
+            expected.id = l[0].id
+            self.assertEqual(l[0], expected)
+
+    def test_ixfr_up_to_date(self):
+        with IXFRNanoNameserver(ixfr_up_to_date_message) as ns:
+            xfr = dns.query.xfr(
+                ns.tcp_address[0],
+                "example",
+                dns.rdatatype.IXFR,
+                port=ns.tcp_address[1],
+                serial=2,
+                relativize=False,
+            )
+            l = list(xfr)
+            self.assertEqual(len(l), 1)
+            expected = dns.message.from_text(
+                ixfr_up_to_date_message, one_rr_per_rrset=True
+            )
+            expected.id = l[0].id
+            self.assertEqual(l[0], expected)
+
+    def test_ixfr_trailing_junk(self):
+        def bad():
+            with IXFRNanoNameserver(ixfr_trailing_junk) as ns:
+                xfr = dns.query.xfr(
+                    ns.tcp_address[0],
+                    "example",
+                    dns.rdatatype.IXFR,
+                    port=ns.tcp_address[1],
+                    serial=2,
+                    relativize=False,
+                )
+                l = list(xfr)
+
+        self.assertRaises(dns.exception.FormError, bad)
+
+    def test_ixfr_base_serial_mismatch(self):
+        def bad():
+            with IXFRNanoNameserver(ixfr_message) as ns:
+                xfr = dns.query.xfr(
+                    ns.tcp_address[0],
+                    "example",
+                    dns.rdatatype.IXFR,
+                    port=ns.tcp_address[1],
+                    serial=1,
+                    relativize=False,
+                )
+                l = list(xfr)
+
+        self.assertRaises(dns.exception.FormError, bad)
+
+
+class TSIGNanoNameserver(Server):
+    def handle(self, request):
+        response = dns.message.make_response(request.message)
+        response.set_rcode(dns.rcode.REFUSED)
+        response.flags |= dns.flags.RA
+        try:
+            if request.qtype == dns.rdatatype.A and request.qclass == dns.rdataclass.IN:
+                rrs = dns.rrset.from_text(request.qname, 300, "IN", "A", "1.2.3.4")
+                response.answer.append(rrs)
+                response.set_rcode(dns.rcode.NOERROR)
+                response.flags |= dns.flags.AA
+        except Exception:
+            pass
+        return response
+
+
+@unittest.skipIf(not _nanonameserver_available, "nanonameserver required")
+class TsigTests(unittest.TestCase):
+    def test_tsig(self):
+        with TSIGNanoNameserver(keyring=keyring) as ns:
+            qname = dns.name.from_text("example.com")
+            q = dns.message.make_query(qname, "A")
+            q.use_tsig(keyring=keyring, keyname="name")
+            response = dns.query.udp(q, ns.udp_address[0], port=ns.udp_address[1])
+            self.assertTrue(response.had_tsig)
+            rrs = response.get_rrset(
+                response.answer, qname, dns.rdataclass.IN, dns.rdatatype.A
+            )
+            self.assertTrue(rrs is not None)
+            seen = set([rdata.address for rdata in rrs])
+            self.assertTrue("1.2.3.4" in seen)
+
+
+@unittest.skipIf(sys.platform == "win32", "low level tests do not work on win32")
+class LowLevelWaitTests(unittest.TestCase):
+    def test_wait_for(self):
+        try:
+            (l, r) = socket.socketpair()
+            # already expired
+            with self.assertRaises(dns.exception.Timeout):
+                dns.query._wait_for(l, True, True, True, 0)
+            # simple timeout
+            with self.assertRaises(dns.exception.Timeout):
+                dns.query._wait_for(l, False, False, False, time.time() + 0.05)
+            # writable no timeout (not hanging is passing)
+            dns.query._wait_for(l, False, True, False, None)
+        finally:
+            l.close()
+            r.close()
+
+
+class MiscTests(unittest.TestCase):
+    def test_matches_destination(self):
+        self.assertTrue(
+            dns.query._matches_destination(
+                socket.AF_INET, ("10.0.0.1", 1234), ("10.0.0.1", 1234), True
+            )
+        )
+        self.assertTrue(
+            dns.query._matches_destination(
+                socket.AF_INET6, ("1::2", 1234), ("0001::2", 1234), True
+            )
+        )
+        self.assertTrue(
+            dns.query._matches_destination(
+                socket.AF_INET, ("10.0.0.1", 1234), None, True
+            )
+        )
+        self.assertFalse(
+            dns.query._matches_destination(
+                socket.AF_INET, ("10.0.0.1", 1234), ("10.0.0.2", 1234), True
+            )
+        )
+        self.assertFalse(
+            dns.query._matches_destination(
+                socket.AF_INET, ("10.0.0.1", 1234), ("10.0.0.1", 1235), True
+            )
+        )
+        with self.assertRaises(dns.query.UnexpectedSource):
+            dns.query._matches_destination(
+                socket.AF_INET, ("10.0.0.1", 1234), ("10.0.0.1", 1235), False
+            )
+
+
+@contextlib.contextmanager
+def mock_udp_recv(wire1, from1, wire2, from2):
+    data = {
+        "saved": dns.query._udp_recv,
+        "first_time": True,
+    }
+
+    def mock(sock, max_size, expiration):
+        if data["first_time"]:
+            data["first_time"] = False
+            return wire1, from1
+        else:
+            return wire2, from2
+
+    try:
+        dns.query._udp_recv = mock
+        yield None
+    finally:
+        dns.query._udp_recv = data["saved"]
+
+
+class MockSock:
+    def __init__(self):
+        self.family = socket.AF_INET
+
+    def sendto(self, data, where):
+        return len(data)
+
+    def close(self):
+        pass
+
+    def setblocking(self, *args):
+        pass
+
+
+class IgnoreErrors(unittest.TestCase):
+    def setUp(self):
+        self.q = dns.message.make_query("example.", "A")
+        self.good_r = dns.message.make_response(self.q)
+        self.good_r.set_rcode(dns.rcode.NXDOMAIN)
+        self.good_r_wire = self.good_r.to_wire()
+
+    def mock_receive(
+        self,
+        wire1,
+        from1,
+        wire2,
+        from2,
+        ignore_unexpected=True,
+        ignore_errors=True,
+        good_r=None,
+    ):
+        if good_r is None:
+            good_r = self.good_r
+        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        try:
+            with mock_udp_recv(wire1, from1, wire2, from2):
+                (r, when) = dns.query.receive_udp(
+                    s,
+                    ("127.0.0.1", 53),
+                    time.time() + 2,
+                    ignore_unexpected=ignore_unexpected,
+                    ignore_errors=ignore_errors,
+                    query=self.q,
+                )
+                self.assertEqual(r, good_r)
+        finally:
+            s.close()
+
+    def test_good_mock(self):
+        self.mock_receive(self.good_r_wire, ("127.0.0.1", 53), None, None)
+
+    def test_bad_address(self):
+        self.mock_receive(
+            self.good_r_wire, ("127.0.0.2", 53), self.good_r_wire, ("127.0.0.1", 53)
+        )
+
+    def test_bad_address_not_ignored(self):
+        def bad():
+            self.mock_receive(
+                self.good_r_wire,
+                ("127.0.0.2", 53),
+                self.good_r_wire,
+                ("127.0.0.1", 53),
+                ignore_unexpected=False,
+            )
+
+        self.assertRaises(dns.query.UnexpectedSource, bad)
+
+    def test_bad_id(self):
+        bad_r = dns.message.make_response(self.q)
+        bad_r.id += 1
+        bad_r_wire = bad_r.to_wire()
+        self.mock_receive(
+            bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+        )
+
+    def test_bad_id_not_ignored(self):
+        bad_r = dns.message.make_response(self.q)
+        bad_r.id += 1
+        bad_r_wire = bad_r.to_wire()
+
+        def bad():
+            (r, wire) = self.mock_receive(
+                bad_r_wire,
+                ("127.0.0.1", 53),
+                self.good_r_wire,
+                ("127.0.0.1", 53),
+                ignore_errors=False,
+            )
+
+        self.assertRaises(AssertionError, bad)
+
+    def test_not_response_not_ignored_udp_level(self):
+        def bad():
+            bad_r = dns.message.make_response(self.q)
+            bad_r.id += 1
+            bad_r_wire = bad_r.to_wire()
+            with mock_udp_recv(
+                bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+            ):
+                s = MockSock()
+                dns.query.udp(self.good_r, "127.0.0.1", sock=s)
+
+        self.assertRaises(dns.query.BadResponse, bad)
+
+    def test_bad_wire(self):
+        bad_r = dns.message.make_response(self.q)
+        bad_r.id += 1
+        bad_r_wire = bad_r.to_wire()
+        self.mock_receive(
+            bad_r_wire[:10], ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+        )
+
+    def test_good_wire_with_truncation_flag_and_no_truncation_raise(self):
+        tc_r = dns.message.make_response(self.q)
+        tc_r.flags |= dns.flags.TC
+        tc_r_wire = tc_r.to_wire()
+        self.mock_receive(tc_r_wire, ("127.0.0.1", 53), None, None, good_r=tc_r)
+
+    def test_wrong_id_wire_with_truncation_flag_and_no_truncation_raise(self):
+        bad_r = dns.message.make_response(self.q)
+        bad_r.id += 1
+        bad_r.flags |= dns.flags.TC
+        bad_r_wire = bad_r.to_wire()
+        self.mock_receive(
+            bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+        )
+
+    def test_bad_wire_not_ignored(self):
+        bad_r = dns.message.make_response(self.q)
+        bad_r.id += 1
+        bad_r_wire = bad_r.to_wire()
+
+        def bad():
+            self.mock_receive(
+                bad_r_wire[:10],
+                ("127.0.0.1", 53),
+                self.good_r_wire,
+                ("127.0.0.1", 53),
+                ignore_errors=False,
+            )
+
+        self.assertRaises(dns.message.ShortHeader, bad)
+
+    def test_trailing_wire(self):
+        wire = self.good_r_wire + b"abcd"
+        self.mock_receive(wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53))
+
+    def test_trailing_wire_not_ignored(self):
+        wire = self.good_r_wire + b"abcd"
+
+        def bad():
+            self.mock_receive(
+                wire,
+                ("127.0.0.1", 53),
+                self.good_r_wire,
+                ("127.0.0.1", 53),
+                ignore_errors=False,
+            )
+
+        self.assertRaises(dns.message.TrailingJunk, bad)
Index: dnspython-1.12.0/dns/inet.py
===================================================================
--- dnspython-1.12.0.orig/dns/inet.py
+++ dnspython-1.12.0/dns/inet.py
@@ -101,11 +101,11 @@ def is_multicast(text):
     @rtype: bool
     """
     try:
-        first = ord(dns.ipv4.inet_aton(text)[0])
+        first = dns.ipv4.inet_aton(text)[0]
         return (first >= 224 and first <= 239)
     except:
         try:
-            first = ord(dns.ipv6.inet_aton(text)[0])
+            first = dns.ipv6.inet_aton(text)[0]
             return (first == 255)
         except:
             raise ValueError

Places

File CVE-2023-29483.patch of Package python-dnspython (Revision 7cfcc7fe3abba23fc2110b8c6294f9e8)

Places