ALP-pool/python-dnspython
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
python-dnspython/CVE-2023-29483.patch

839 lines
28 KiB
Diff
Raw Normal View History

Sync from SUSE:ALP:Source:Standard:1.0 python-dnspython revision cb1a11be65bea4af6f9b21d0640bb9bb
2025-01-13 14:35:25 +01:00
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-2.4.2/dns/asyncquery.py
===================================================================
--- dnspython-2.4.2.orig/dns/asyncquery.py
+++ dnspython-2.4.2/dns/asyncquery.py
@@ -120,6 +120,8 @@ async def receive_udp(
request_mac: Optional[bytes] = b"",
ignore_trailing: bool = False,
raise_on_truncation: bool = False,
+ ignore_errors: bool = False,
+ query: Optional[dns.message.Message] = None,
) -> Any:
"""Read a DNS message from a UDP socket.
@@ -133,22 +135,40 @@ async def receive_udp(
"""
wire = b""
- while 1:
+ while True:
(wire, from_address) = await sock.recvfrom(65535, _timeout(expiration))
- if _matches_destination(
+ if not _matches_destination(
sock.family, from_address, destination, ignore_unexpected
):
- break
- received_time = time.time()
- r = dns.message.from_wire(
- wire,
- keyring=keyring,
- request_mac=request_mac,
- one_rr_per_rrset=one_rr_per_rrset,
- ignore_trailing=ignore_trailing,
- raise_on_truncation=raise_on_truncation,
- )
- return (r, received_time, from_address)
+ 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,
+ ignore_trailing=ignore_trailing,
+ raise_on_truncation=raise_on_truncation,
+ )
+ except dns.message.Truncated as e:
+ # See the comment in query.py for details.
+ if (
+ ignore_errors
+ and query is not None
+ and not query.is_response(e.message())
+ ):
+ continue
+ else:
+ raise
+ 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, from_address)
async def udp(
@@ -164,6 +184,7 @@ async def udp(
raise_on_truncation: bool = False,
sock: Optional[dns.asyncbackend.DatagramSocket] = None,
backend: Optional[dns.asyncbackend.Backend] = None,
+ ignore_errors: bool = False,
) -> dns.message.Message:
"""Return the response obtained after sending a query via UDP.
@@ -205,9 +226,13 @@ async def udp(
q.mac,
ignore_trailing,
raise_on_truncation,
+ ignore_errors,
+ q,
)
r.time = received_time - begin_time
- 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
@@ -225,6 +250,7 @@ async def udp_with_fallback(
udp_sock: Optional[dns.asyncbackend.DatagramSocket] = None,
tcp_sock: Optional[dns.asyncbackend.StreamSocket] = None,
backend: Optional[dns.asyncbackend.Backend] = None,
+ ignore_errors: bool = False,
) -> Tuple[dns.message.Message, bool]:
"""Return the response to the query, trying UDP first and falling back
to TCP if UDP results in a truncated response.
@@ -260,6 +286,7 @@ async def udp_with_fallback(
True,
udp_sock,
backend,
+ ignore_errors,
)
return (response, False)
except dns.message.Truncated:
Index: dnspython-2.4.2/dns/nameserver.py
===================================================================
--- dnspython-2.4.2.orig/dns/nameserver.py
+++ dnspython-2.4.2/dns/nameserver.py
@@ -115,6 +115,7 @@ class Do53Nameserver(AddressAndPortNames
raise_on_truncation=True,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
+ ignore_errors=True,
)
return response
@@ -153,6 +154,7 @@ class Do53Nameserver(AddressAndPortNames
backend=backend,
one_rr_per_rrset=one_rr_per_rrset,
ignore_trailing=ignore_trailing,
+ ignore_errors=True,
)
return response
Index: dnspython-2.4.2/dns/query.py
===================================================================
--- dnspython-2.4.2.orig/dns/query.py
+++ dnspython-2.4.2/dns/query.py
@@ -577,6 +577,8 @@ def receive_udp(
request_mac: Optional[bytes] = b"",
ignore_trailing: bool = False,
raise_on_truncation: bool = False,
+ ignore_errors: bool = False,
+ query: Optional[dns.message.Message] = None,
) -> Any:
"""Read a DNS message from a UDP socket.
@@ -617,28 +619,58 @@ def receive_udp(
``(dns.message.Message, float, tuple)``
tuple of the received message, the received time, and the address where
the message arrived from.
+
+ *ignore_errors*, a ``bool``. If various format errors or response
+ mismatches occur, ignore them and keep listening for a valid response.
+ The default is ``False``.
+
+ *query*, a ``dns.message.Message`` or ``None``. 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.
"""
wire = b""
while True:
(wire, from_address) = _udp_recv(sock, 65535, expiration)
- if _matches_destination(
+ if not _matches_destination(
sock.family, from_address, destination, ignore_unexpected
):
- break
- received_time = time.time()
- r = dns.message.from_wire(
- wire,
- keyring=keyring,
- request_mac=request_mac,
- one_rr_per_rrset=one_rr_per_rrset,
- ignore_trailing=ignore_trailing,
- raise_on_truncation=raise_on_truncation,
- )
- if destination:
- return (r, received_time)
- else:
- return (r, received_time, from_address)
+ 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,
+ ignore_trailing=ignore_trailing,
+ raise_on_truncation=raise_on_truncation,
+ )
+ except dns.message.Truncated as e:
+ # If we got Truncated and not FORMERR, we at least got the header with TC
+ # set, and very likely the question section, so we'll re-raise if the
+ # message seems to be a response as we need to know when truncation happens.
+ # We need to check that it seems to be a response as we don't want a random
+ # injected message with TC set to cause us to bail out.
+ if (
+ ignore_errors
+ and query is not None
+ and not query.is_response(e.message())
+ ):
+ continue
+ else:
+ raise
+ except Exception:
+ if ignore_errors:
+ continue
+ else:
+ raise
+ if ignore_errors and query is not None and not query.is_response(r):
+ continue
+ if destination:
+ return (r, received_time)
+ else:
+ return (r, received_time, from_address)
def udp(
@@ -653,6 +685,7 @@ def udp(
ignore_trailing: bool = False,
raise_on_truncation: bool = False,
sock: Optional[Any] = None,
+ ignore_errors: bool = False,
) -> dns.message.Message:
"""Return the response obtained after sending a query via UDP.
@@ -689,6 +722,10 @@ def udp(
if a socket is provided, it must be a nonblocking datagram socket,
and the *source* and *source_port* are ignored.
+ *ignore_errors*, a ``bool``. If various format errors or response
+ mismatches occur, ignore them and keep listening for a valid response.
+ The default is ``False``.
+
Returns a ``dns.message.Message``.
"""
@@ -713,9 +750,13 @@ def udp(
q.mac,
ignore_trailing,
raise_on_truncation,
+ ignore_errors,
+ q,
)
r.time = received_time - begin_time
- 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
assert (
@@ -735,48 +776,50 @@ def udp_with_fallback(
ignore_trailing: bool = False,
udp_sock: Optional[Any] = None,
tcp_sock: Optional[Any] = None,
+ ignore_errors: bool = False,
) -> Tuple[dns.message.Message, bool]:
"""Return the response to the query, trying UDP first and falling back
to TCP if UDP results in a truncated response.
*q*, a ``dns.message.Message``, the query to send
- *where*, a ``str`` containing an IPv4 or IPv6 address, where
- to send the message.
+ *where*, a ``str`` containing an IPv4 or IPv6 address, where to send the message.
- *timeout*, a ``float`` or ``None``, the number of seconds to wait before the
- query times out. If ``None``, the default, wait forever.
+ *timeout*, a ``float`` or ``None``, the number of seconds to wait before the query
+ times out. If ``None``, the default, wait forever.
*port*, an ``int``, the port send the message to. The default is 53.
- *source*, a ``str`` containing an IPv4 or IPv6 address, specifying
- the source address. The default is the wildcard address.
+ *source*, a ``str`` containing an IPv4 or IPv6 address, specifying the source
+ address. The default is the wildcard address.
- *source_port*, an ``int``, the port from which to send the message.
- The default is 0.
+ *source_port*, an ``int``, the port from which to send the message. The default is
+ 0.
- *ignore_unexpected*, a ``bool``. If ``True``, ignore responses from
- unexpected sources.
+ *ignore_unexpected*, a ``bool``. If ``True``, ignore responses from unexpected
+ sources.
- *one_rr_per_rrset*, a ``bool``. If ``True``, put each RR into its own
- RRset.
+ *one_rr_per_rrset*, a ``bool``. If ``True``, put each RR into its own RRset.
- *ignore_trailing*, a ``bool``. If ``True``, ignore trailing
- junk at end of the received message.
+ *ignore_trailing*, a ``bool``. If ``True``, ignore trailing junk at end of the
+ received message.
- *udp_sock*, a ``socket.socket``, or ``None``, the socket to use for the
- UDP query. If ``None``, the default, a socket is created. Note that
- if a socket is provided, it must be a nonblocking datagram socket,
- and the *source* and *source_port* are ignored for the UDP query.
+ *udp_sock*, a ``socket.socket``, or ``None``, the socket to use for the UDP query.
+ If ``None``, the default, a socket is created. Note that if a socket is provided,
+ it must be a nonblocking datagram socket, and the *source* and *source_port* are
+ ignored for the UDP query.
*tcp_sock*, a ``socket.socket``, or ``None``, the connected socket to use for the
- TCP query. If ``None``, the default, a socket is created. Note that
- if a socket is provided, it must be a nonblocking connected stream
- socket, and *where*, *source* and *source_port* are ignored for the TCP
- query.
+ TCP query. If ``None``, the default, a socket is created. Note that if a socket is
+ provided, it must be a nonblocking connected stream socket, and *where*, *source*
+ and *source_port* are ignored for the TCP query.
+
+ *ignore_errors*, a ``bool``. If various format errors or response mismatches occur
+ while listening for UDP, ignore them and keep listening for a valid response. The
+ default is ``False``.
- Returns a (``dns.message.Message``, tcp) tuple where tcp is ``True``
- if and only if TCP was used.
+ Returns a (``dns.message.Message``, tcp) tuple where tcp is ``True`` if and only if
+ TCP was used.
"""
try:
response = udp(
@@ -791,6 +834,7 @@ def udp_with_fallback(
ignore_trailing,
True,
udp_sock,
+ ignore_errors,
)
return (response, False)
except dns.message.Truncated:
Index: dnspython-2.4.2/tests/test_query.py
===================================================================
--- dnspython-2.4.2.orig/tests/test_query.py
+++ dnspython-2.4.2/tests/test_query.py
@@ -15,6 +15,7 @@
# 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
@@ -28,10 +29,12 @@ 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
@@ -659,3 +662,203 @@ class MiscTests(unittest.TestCase):
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):
+ saved = dns.query._udp_recv
+ first_time = True
+
+ def mock(sock, max_size, expiration):
+ nonlocal first_time
+ if first_time:
+ first_time = False
+ return wire1, from1
+ else:
+ return wire2, from2
+
+ try:
+ dns.query._udp_recv = mock
+ yield None
+ finally:
+ dns.query._udp_recv = saved
+
+
+class MockSock:
+ def __init__(self):
+ self.family = socket.AF_INET
+
+ def sendto(self, data, where):
+ return len(data)
+
+
+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,
+ raise_on_truncation=False,
+ 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,
+ raise_on_truncation=raise_on_truncation,
+ 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_good_wire_with_truncation_flag_and_truncation_raise(self):
+ def good():
+ 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, raise_on_truncation=True
+ )
+
+ self.assertRaises(dns.message.Truncated, good)
+
+ 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_wrong_id_wire_with_truncation_flag_and_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),
+ raise_on_truncation=True
+ )
+
+ 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-2.4.2/tests/test_async.py
===================================================================
--- dnspython-2.4.2.orig/tests/test_async.py
+++ dnspython-2.4.2/tests/test_async.py
@@ -15,10 +15,10 @@
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+import sys
import asyncio
import random
import socket
-import sys
import time
import unittest
@@ -28,6 +28,7 @@ import dns.asyncresolver
import dns.message
import dns.name
import dns.query
+import dns.rcode
import dns.rdataclass
import dns.rdatatype
import dns.resolver
@@ -688,3 +689,243 @@ try:
except ImportError:
pass
+
+
+class MockSock:
+ def __init__(self, wire1, from1, wire2, from2):
+ self.family = socket.AF_INET
+ self.first_time = True
+ self.wire1 = wire1
+ self.from1 = from1
+ self.wire2 = wire2
+ self.from2 = from2
+
+ async def sendto(self, data, where, timeout):
+ return len(data)
+
+ async def recvfrom(self, bufsize, expiration):
+ if self.first_time:
+ self.first_time = False
+ return self.wire1, self.from1
+ else:
+ return self.wire2, self.from2
+
+
+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()
+ dns.asyncbackend.set_default_backend("asyncio")
+
+ def async_run(self, afunc):
+ return asyncio.run(afunc())
+
+ async def mock_receive(
+ self,
+ wire1,
+ from1,
+ wire2,
+ from2,
+ ignore_unexpected=True,
+ ignore_errors=True,
+ raise_on_truncation=False,
+ good_r=None,
+ ):
+ if good_r is None:
+ good_r = self.good_r
+ s = MockSock(wire1, from1, wire2, from2)
+ (r, when, _) = await dns.asyncquery.receive_udp(
+ s,
+ ("127.0.0.1", 53),
+ time.time() + 2,
+ ignore_unexpected=ignore_unexpected,
+ ignore_errors=ignore_errors,
+ raise_on_truncation=raise_on_truncation,
+ query=self.q,
+ )
+ self.assertEqual(r, good_r)
+
+ def test_good_mock(self):
+ async def run():
+ await self.mock_receive(self.good_r_wire, ("127.0.0.1", 53), None, None)
+
+ self.async_run(run)
+
+ def test_bad_address(self):
+ async def run():
+ await self.mock_receive(
+ self.good_r_wire, ("127.0.0.2", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+
+ self.async_run(run)
+
+ def test_bad_address_not_ignored(self):
+ async def abad():
+ await self.mock_receive(
+ self.good_r_wire,
+ ("127.0.0.2", 53),
+ self.good_r_wire,
+ ("127.0.0.1", 53),
+ ignore_unexpected=False,
+ )
+
+ def bad():
+ self.async_run(abad)
+
+ self.assertRaises(dns.query.UnexpectedSource, bad)
+
+ def test_not_response_not_ignored_udp_level(self):
+ async def abad():
+ bad_r = dns.message.make_response(self.q)
+ bad_r.id += 1
+ bad_r_wire = bad_r.to_wire()
+ s = MockSock(
+ bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+ await dns.asyncquery.udp(self.good_r, "127.0.0.1", sock=s)
+
+ def bad():
+ self.async_run(abad)
+
+ self.assertRaises(dns.query.BadResponse, bad)
+
+ def test_bad_id(self):
+ async def run():
+ bad_r = dns.message.make_response(self.q)
+ bad_r.id += 1
+ bad_r_wire = bad_r.to_wire()
+ await self.mock_receive(
+ bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+
+ self.async_run(run)
+
+ 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()
+
+ async def abad():
+ (r, wire) = await self.mock_receive(
+ bad_r_wire,
+ ("127.0.0.1", 53),
+ self.good_r_wire,
+ ("127.0.0.1", 53),
+ ignore_errors=False,
+ )
+
+ def bad():
+ self.async_run(abad)
+
+ self.assertRaises(AssertionError, bad)
+
+ def test_bad_wire(self):
+ async def run():
+ bad_r = dns.message.make_response(self.q)
+ bad_r.id += 1
+ bad_r_wire = bad_r.to_wire()
+ await self.mock_receive(
+ bad_r_wire[:10], ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+
+ self.async_run(run)
+
+ def test_good_wire_with_truncation_flag_and_no_truncation_raise(self):
+ async def run():
+ tc_r = dns.message.make_response(self.q)
+ tc_r.flags |= dns.flags.TC
+ tc_r_wire = tc_r.to_wire()
+ await self.mock_receive(
+ tc_r_wire, ("127.0.0.1", 53), None, None, good_r=tc_r
+ )
+
+ self.async_run(run)
+
+ def test_good_wire_with_truncation_flag_and_truncation_raise(self):
+ async def agood():
+ tc_r = dns.message.make_response(self.q)
+ tc_r.flags |= dns.flags.TC
+ tc_r_wire = tc_r.to_wire()
+ await self.mock_receive(
+ tc_r_wire, ("127.0.0.1", 53), None, None, raise_on_truncation=True
+ )
+
+ def good():
+ self.async_run(agood)
+
+ self.assertRaises(dns.message.Truncated, good)
+
+ def test_wrong_id_wire_with_truncation_flag_and_no_truncation_raise(self):
+ async def run():
+ 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()
+ await self.mock_receive(
+ bad_r_wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+
+ self.async_run(run)
+
+ def test_wrong_id_wire_with_truncation_flag_and_truncation_raise(self):
+ async def run():
+ 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()
+ await self.mock_receive(
+ bad_r_wire,
+ ("127.0.0.1", 53),
+ self.good_r_wire,
+ ("127.0.0.1", 53),
+ raise_on_truncation=True,
+ )
+
+ self.async_run(run)
+
+ 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()
+
+ async def abad():
+ await self.mock_receive(
+ bad_r_wire[:10],
+ ("127.0.0.1", 53),
+ self.good_r_wire,
+ ("127.0.0.1", 53),
+ ignore_errors=False,
+ )
+
+ def bad():
+ self.async_run(abad)
+
+ self.assertRaises(dns.message.ShortHeader, bad)
+
+ def test_trailing_wire(self):
+ async def run():
+ wire = self.good_r_wire + b"abcd"
+ await self.mock_receive(
+ wire, ("127.0.0.1", 53), self.good_r_wire, ("127.0.0.1", 53)
+ )
+
+ self.async_run(run)
+
+ def test_trailing_wire_not_ignored(self):
+ wire = self.good_r_wire + b"abcd"
+
+ async def abad():
+ await self.mock_receive(
+ wire,
+ ("127.0.0.1", 53),
+ self.good_r_wire,
+ ("127.0.0.1", 53),
+ ignore_errors=False,
+ )
+
+ def bad():
+ self.async_run(abad)
+
+ self.assertRaises(dns.message.TrailingJunk, bad)
Reference in New Issue Copy Permalink