1 #!/usr/bin/python3 2 3 # Copyright (C) Internet Systems Consortium, Inc. ("ISC") 4 # 5 # SPDX-License-Identifier: MPL-2.0 6 # 7 # This Source Code Form is subject to the terms of the Mozilla Public 8 # License, v. 2.0. If a copy of the MPL was not distributed with this 9 # file, you can obtain one at https://mozilla.org/MPL/2.0/. 10 # 11 # See the COPYRIGHT file distributed with this work for additional 12 # information regarding copyright ownership. 13 14 # pylint: disable=unused-variable 15 16 import socket 17 import time 18 19 import pytest 20 21 pytest.importorskip("dns", minversion="2.0.0") 22 import dns.edns 23 import dns.message 24 import dns.name 25 import dns.query 26 import dns.rdataclass 27 import dns.rdatatype 28 29 import isctest.mark # pylint: disable=import-error 30 31 pytestmark = pytest.mark.extra_artifacts( 32 [ 33 "ns1/large.db", 34 ] 35 ) 36 37 TIMEOUT = 10 38 39 40 def create_msg(qname, qtype): 41 msg = dns.message.make_query( 42 qname, qtype, want_dnssec=True, use_edns=0, payload=4096 43 ) 44 return msg 45 46 47 def timeout(): 48 return time.time() + TIMEOUT 49 50 51 def test_initial_timeout(named_port): 52 # 53 # The initial timeout is 2.5 seconds, so this should timeout 54 # 55 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 56 sock.connect(("10.53.0.1", named_port)) 57 58 time.sleep(3) 59 60 msg = create_msg("example.", "A") 61 62 with pytest.raises(EOFError): 63 try: 64 dns.query.send_tcp(sock, msg, timeout()) 65 dns.query.receive_tcp(sock, timeout()) 66 except ConnectionError as e: 67 raise EOFError from e 68 69 70 @pytest.mark.flaky(max_runs=2, rerun_filter=isctest.mark.is_host_freebsd_13) 71 def test_idle_timeout(named_port): 72 # 73 # The idle timeout is 5 seconds, so the third message should fail 74 # 75 msg = create_msg("example.", "A") 76 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 77 sock.connect(("10.53.0.1", named_port)) 78 79 time.sleep(1) 80 81 dns.query.send_tcp(sock, msg, timeout()) 82 dns.query.receive_tcp(sock, timeout()) 83 84 time.sleep(2) 85 86 dns.query.send_tcp(sock, msg, timeout()) 87 dns.query.receive_tcp(sock, timeout()) 88 89 time.sleep(6) 90 91 with pytest.raises(EOFError): 92 try: 93 dns.query.send_tcp(sock, msg, timeout()) 94 dns.query.receive_tcp(sock, timeout()) 95 except ConnectionError as e: 96 raise EOFError from e 97 98 99 def test_keepalive_timeout(named_port): 100 # 101 # Keepalive is 7 seconds, so the third message should succeed. 102 # 103 msg = create_msg("example.", "A") 104 kopt = dns.edns.GenericOption(11, b"\x00") 105 msg.use_edns(edns=True, payload=4096, options=[kopt]) 106 107 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 108 sock.connect(("10.53.0.1", named_port)) 109 110 time.sleep(1) 111 112 dns.query.send_tcp(sock, msg, timeout()) 113 dns.query.receive_tcp(sock, timeout()) 114 115 time.sleep(2) 116 117 dns.query.send_tcp(sock, msg, timeout()) 118 dns.query.receive_tcp(sock, timeout()) 119 120 time.sleep(6) 121 122 dns.query.send_tcp(sock, msg, timeout()) 123 dns.query.receive_tcp(sock, timeout()) 124 125 126 def test_pipelining_timeout(named_port): 127 # 128 # The pipelining should only timeout after the last message is received 129 # 130 msg = create_msg("example.", "A") 131 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 132 sock.connect(("10.53.0.1", named_port)) 133 134 time.sleep(1) 135 136 # Send and receive 25 DNS queries 137 for _ in range(25): 138 dns.query.send_tcp(sock, msg, timeout()) 139 for _ in range(25): 140 dns.query.receive_tcp(sock, timeout()) 141 142 time.sleep(3) 143 144 # Send and receive 25 DNS queries 145 for _ in range(25): 146 dns.query.send_tcp(sock, msg, timeout()) 147 for _ in range(25): 148 dns.query.receive_tcp(sock, timeout()) 149 150 time.sleep(6) 151 152 with pytest.raises(EOFError): 153 try: 154 dns.query.send_tcp(sock, msg, timeout()) 155 dns.query.receive_tcp(sock, timeout()) 156 except ConnectionError as e: 157 raise EOFError from e 158 159 160 def test_long_axfr(named_port): 161 # 162 # The timers should not fire during AXFR, thus the connection should not 163 # close abruptly 164 # 165 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 166 sock.connect(("10.53.0.1", named_port)) 167 168 name = dns.name.from_text("example.") 169 msg = create_msg("example.", "AXFR") 170 dns.query.send_tcp(sock, msg, timeout()) 171 172 # Receive the initial DNS message with SOA 173 (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True) 174 soa = response.get_rrset( 175 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 176 ) 177 assert soa is not None 178 179 # Pull DNS message from wire until the second SOA is received 180 while True: 181 (response, _) = dns.query.receive_tcp( 182 sock, timeout(), one_rr_per_rrset=True 183 ) 184 soa = response.get_rrset( 185 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 186 ) 187 if soa is not None: 188 break 189 assert soa is not None 190 191 192 @pytest.mark.flaky(max_runs=3) 193 def test_send_timeout(named_port): 194 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 195 sock.connect(("10.53.0.1", named_port)) 196 197 # Send and receive single large RDATA over TCP 198 msg = create_msg("large.example.", "TXT") 199 dns.query.send_tcp(sock, msg, timeout()) 200 dns.query.receive_tcp(sock, timeout()) 201 202 # Send and receive 28 large (~32k) DNS queries that should 203 # fill the default maximum 208k TCP send buffer 204 for _ in range(28): 205 dns.query.send_tcp(sock, msg, timeout()) 206 207 # configure idle interval is 5 seconds, sleep 6 to make sure we are 208 # above the interval 209 time.sleep(6) 210 211 with pytest.raises(EOFError): 212 try: 213 dns.query.send_tcp(sock, msg, timeout()) 214 dns.query.receive_tcp(sock, timeout()) 215 except ConnectionError as e: 216 raise EOFError from e 217 218 219 @isctest.mark.long_test 220 def test_max_transfer_idle_out(named_port): 221 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 222 sock.connect(("10.53.0.1", named_port)) 223 224 name = dns.name.from_text("example.") 225 msg = create_msg("example.", "AXFR") 226 dns.query.send_tcp(sock, msg, timeout()) 227 228 # Receive the initial DNS message with SOA 229 (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True) 230 soa = response.get_rrset( 231 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 232 ) 233 assert soa is not None 234 235 time.sleep(61) # max-transfer-idle-out is 1 minute 236 237 with pytest.raises(ConnectionResetError): 238 # Process queued TCP messages 239 while True: 240 (response, _) = dns.query.receive_tcp( 241 sock, timeout(), one_rr_per_rrset=True 242 ) 243 soa = response.get_rrset( 244 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 245 ) 246 if soa is not None: 247 break 248 assert soa is None 249 250 251 @isctest.mark.long_test 252 def test_max_transfer_time_out(named_port): 253 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 254 sock.connect(("10.53.0.1", named_port)) 255 256 name = dns.name.from_text("example.") 257 msg = create_msg("example.", "AXFR") 258 dns.query.send_tcp(sock, msg, timeout()) 259 260 # Receive the initial DNS message with SOA 261 (response, _) = dns.query.receive_tcp(sock, timeout(), one_rr_per_rrset=True) 262 soa = response.get_rrset( 263 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 264 ) 265 assert soa is not None 266 267 # The loop should timeout at the 5 minutes (max-transfer-time-out) 268 with pytest.raises(EOFError): 269 while True: 270 time.sleep(1) 271 (response, _) = dns.query.receive_tcp( 272 sock, timeout(), one_rr_per_rrset=True 273 ) 274 soa = response.get_rrset( 275 dns.message.ANSWER, name, dns.rdataclass.IN, dns.rdatatype.SOA 276 ) 277 if soa is not None: 278 break 279 assert soa is None 280
Indexes created Sun Mar 01 05:31:48 UTC 2026