1 # Copyright (C) Internet Systems Consortium, Inc. ("ISC") 2 # 3 # SPDX-License-Identifier: MPL-2.0 4 # 5 # This Source Code Form is subject to the terms of the Mozilla Public 6 # License, v. 2.0. If a copy of the MPL was not distributed with this 7 # file, you can obtain one at https://mozilla.org/MPL/2.0/. 8 # 9 # See the COPYRIGHT file distributed with this work for additional 10 # information regarding copyright ownership. 11 12 from typing import SupportsInt 13 14 import socket 15 16 import dns.flags 17 import dns.name 18 import dns.opcode 19 import dns.rcode 20 import dns.rdataclass 21 import dns.rdatatype 22 import dns.rdtypes.ANY.NSEC3 23 import dns.rdtypes.ANY.OPT 24 import dns.rdtypes.ANY.RRSIG 25 import dns.rdtypes.ANY.SOA 26 import dns.rdtypes.ANY.TSIG 27 import dns.rdtypes.IN.A 28 import pytest 29 30 31 def wire(*parts: bytes) -> bytes: 32 return b"".join(parts) 33 34 35 def u16(value: SupportsInt) -> bytes: 36 return int(value).to_bytes(2, byteorder="big") 37 38 39 def u32(value: SupportsInt) -> bytes: 40 return int(value).to_bytes(4, byteorder="big") 41 42 43 def name(text: str) -> bytes: 44 return dns.name.from_text(text).to_wire() 45 46 47 def name_pointer(offset: int) -> bytes: 48 return u16(0xC000 | offset) 49 50 51 def root() -> bytes: 52 return dns.name.root.to_wire() 53 54 55 def header( 56 message_id: int = 0, 57 flags: int = 0, 58 opcode: int = dns.opcode.QUERY, 59 rcode: dns.rcode.Rcode = dns.rcode.NOERROR, 60 qdcount: int = 0, 61 ancount: int = 0, 62 nscount: int = 0, 63 arcount: int = 0, 64 ) -> bytes: 65 return wire( 66 u16(message_id), 67 u16(flags | dns.opcode.to_flags(opcode) | int(rcode)), 68 u16(qdcount), 69 u16(ancount), 70 u16(nscount), 71 u16(arcount), 72 ) 73 74 75 def formerr_response_header( 76 message_id: int = 0, 77 opcode: int = dns.opcode.QUERY, 78 rcode: dns.rcode.Rcode = dns.rcode.FORMERR, 79 qdcount: int = 0, 80 ancount: int = 0, 81 nscount: int = 0, 82 arcount: int = 0, 83 ) -> bytes: 84 return header( 85 message_id=message_id, 86 flags=dns.flags.QR, 87 opcode=opcode, 88 rcode=rcode, 89 qdcount=qdcount, 90 ancount=ancount, 91 nscount=nscount, 92 arcount=arcount, 93 ) 94 95 96 def question( 97 qname: bytes, 98 qtype: dns.rdatatype.RdataType = dns.rdatatype.RdataType.A, 99 qclass: dns.rdataclass.RdataClass = dns.rdataclass.RdataClass.IN, 100 ) -> bytes: 101 return wire( 102 qname, 103 u16(qtype), 104 u16(qclass), 105 ) 106 107 108 def rr( 109 owner: bytes, 110 rrtype: SupportsInt, 111 rrclass: SupportsInt, 112 *, 113 ttl: int = 1, 114 rdata: bytes = b"", 115 ) -> bytes: 116 return wire( 117 question( 118 owner, dns.rdatatype.RdataType(rrtype), dns.rdataclass.RdataClass(rrclass) 119 ), 120 u32(ttl), 121 u16(len(rdata)), 122 rdata, 123 ) 124 125 126 def oversized_name() -> bytes: 127 labels = [bytes([15]) + b"A" * 15 for _ in range(15)] 128 labels.append(bytes([14]) + b"A" * 14 + root()) 129 return wire(*labels) 130 131 132 def soa_rr( 133 *, 134 minimum: int, 135 ) -> bytes: 136 return rr( 137 root(), 138 dns.rdatatype.RdataType.SOA, 139 dns.rdataclass.RdataClass.IN, 140 rdata=dns.rdtypes.ANY.SOA.SOA( 141 dns.rdataclass.RdataClass.IN, 142 dns.rdatatype.RdataType.SOA, 143 dns.name.root, 144 dns.name.root, 145 1, 146 2, 147 3, 148 4, 149 minimum, 150 ).to_wire(), 151 ) 152 153 154 def nsec3_rr( 155 *, 156 owner: bytes, 157 ) -> bytes: 158 return rr( 159 owner, 160 dns.rdatatype.RdataType.NSEC3, 161 dns.rdataclass.RdataClass.IN, 162 rdata=dns.rdtypes.ANY.NSEC3.NSEC3( 163 dns.rdataclass.RdataClass.IN, 164 dns.rdatatype.RdataType.NSEC3, 165 algorithm=240, 166 flags=0, 167 iterations=0, 168 salt=b"", 169 next=b"\xff", 170 windows=[], 171 ).to_wire(), 172 ) 173 174 175 def key_rdata( 176 *, 177 flags: int, 178 protocol: int, 179 algorithm: int, 180 keydata: bytes, 181 ) -> bytes: 182 # No dns.rdtypes.ANY.KEY.KEY class, so construct the rdata manually 183 return wire(u16(flags), bytes([protocol, algorithm]), keydata) 184 185 186 def key_rr(*, rdclass: dns.rdataclass.RdataClass) -> bytes: 187 return rr( 188 root(), 189 dns.rdatatype.RdataType.KEY, 190 rdclass, 191 rdata=key_rdata( 192 flags=0, 193 protocol=0, 194 algorithm=248, 195 keydata=b"\x00", 196 ), 197 ) 198 199 200 def malformed_rrsig_rr() -> bytes: 201 return rr( 202 root(), 203 dns.rdatatype.RdataType.RRSIG, 204 dns.rdataclass.RdataClass.IN, 205 rdata=dns.rdtypes.ANY.RRSIG.RRSIG( 206 dns.rdataclass.RdataClass.IN, 207 dns.rdatatype.RdataType.RRSIG, 208 0, 209 240, 210 0, 211 1, 212 2, 213 3, 214 0, 215 dns.name.root, 216 b"\x00", 217 ).to_wire(), 218 ) 219 220 221 def tsig_rr( 222 *, 223 owner: bytes = root(), 224 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.RdataClass.ANY, 225 algorithm: dns.name.Name = dns.name.root, 226 time_signed: int = 0x010203040506, 227 fudge: int = 0x0102, 228 mac: bytes = b"\x00", 229 original_id: int = 0, 230 error: int = 0, 231 other: bytes = b"", 232 ) -> bytes: 233 return rr( 234 owner, 235 dns.rdatatype.RdataType.TSIG, 236 rdclass, 237 rdata=dns.rdtypes.ANY.TSIG.TSIG( 238 rdclass, 239 dns.rdatatype.RdataType.TSIG, 240 algorithm, 241 time_signed, 242 fudge, 243 mac, 244 original_id, 245 error, 246 other, 247 ).to_wire(), 248 ) 249 250 251 def opt_rr(*, owner: bytes) -> bytes: 252 return rr( 253 owner, 254 dns.rdatatype.RdataType.OPT, 255 dns.rdataclass.RdataClass.IN, 256 ttl=0, 257 rdata=dns.rdtypes.ANY.OPT.OPT( 258 dns.rdataclass.RdataClass.IN, 259 dns.rdatatype.RdataType.OPT, 260 [], 261 ).to_wire(), 262 ) 263 264 265 def a_rdata(ipv4_bytes: bytes = b"\x00\x00\x00\x00") -> bytes: 266 return dns.rdtypes.IN.A.A( 267 dns.rdataclass.RdataClass.IN, 268 dns.rdatatype.RdataType.A, 269 ipv4_bytes, 270 ).to_wire() 271 272 273 def a_rr(owner: bytes = root()) -> bytes: 274 return rr( 275 owner, 276 dns.rdatatype.RdataType.A, 277 dns.rdataclass.RdataClass.IN, 278 rdata=a_rdata(), 279 ) 280 281 282 def query_raw_tcp(host: str, port: int, packet_wire: bytes) -> bytes: 283 with ( 284 socket.create_connection((host, port), timeout=10) as sock, 285 sock.makefile("rwb") as f, 286 ): 287 f.write(u16(len(packet_wire))) 288 f.write(packet_wire) 289 f.flush() 290 length = int.from_bytes(f.read(2), byteorder="big") 291 return f.read(length) 292 293 294 @pytest.mark.parametrize( 295 "query_wire,expected_wire", 296 [ 297 pytest.param( 298 wire( 299 header(qdcount=1), 300 question(oversized_name()), 301 ), 302 formerr_response_header(), 303 id="nametoolong", 304 ), 305 pytest.param( 306 wire( 307 header(qdcount=2), 308 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.A), 309 # Two names concatenated in the QNAME field 310 question( 311 wire(name("AAAAAAAAAAAAAA."), name("AAAAAAAAAAAAAB.")), 312 dns.rdatatype.RdataType.A, 313 ), 314 ), 315 formerr_response_header(), 316 id="twoquestionnames", 317 ), 318 pytest.param( 319 wire( 320 header(qdcount=2), 321 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.A), 322 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.NS), 323 ), 324 wire( 325 formerr_response_header(qdcount=2), 326 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.A), 327 question(name_pointer(12), dns.rdatatype.RdataType.NS), 328 ), 329 id="twoquestiontypes", 330 ), 331 pytest.param( 332 wire( 333 header(qdcount=2), 334 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.A), 335 question(name("AAAAAAAAAAAAAA."), dns.rdatatype.RdataType.A), 336 ), 337 formerr_response_header(), 338 id="dupquestion", 339 ), 340 pytest.param( 341 wire( 342 header(qdcount=1, ancount=2), 343 question(root(), dns.rdatatype.RdataType.SOA), 344 soa_rr(minimum=5), 345 soa_rr(minimum=6), 346 ), 347 wire( 348 formerr_response_header(qdcount=1), 349 question(root(), dns.rdatatype.RdataType.SOA), 350 ), 351 id="dupans", 352 ), 353 pytest.param( 354 wire( 355 header(ancount=1), 356 rr( 357 root(), 358 dns.rdatatype.RdataType.MAILB, 359 dns.rdataclass.RdataClass.IN, 360 ), 361 ), 362 formerr_response_header(), 363 id="qtypeasanswer", 364 ), 365 pytest.param( 366 header(), 367 # This would be NOERROR if it included a COOKIE option, 368 # but is a FORMERR without one. 369 formerr_response_header(), 370 id="noquestions", 371 ), 372 pytest.param( 373 wire( 374 header(qdcount=1, nscount=1), 375 question(root(), dns.rdatatype.RdataType.A), 376 # Bad NSEC3 owner: X. is not in the base32hex alphabet. 377 nsec3_rr(owner=name("X.")), 378 ), 379 wire( 380 formerr_response_header(rcode=dns.rcode.SERVFAIL, qdcount=1), 381 question(root(), dns.rdatatype.RdataType.A), 382 ), 383 id="badnsec3owner", 384 ), 385 pytest.param( 386 wire( 387 header(arcount=1), 388 # Truncated A record (no ttl, length or data) 389 question(root(), dns.rdatatype.RdataType.A), 390 ), 391 formerr_response_header(), 392 id="shortrecord", 393 ), 394 pytest.param( 395 wire( 396 header(qdcount=1), 397 # Truncated question (no class) 398 root(), 399 u16(dns.rdatatype.RdataType.A), 400 ), 401 formerr_response_header(), 402 id="shortquestion", 403 ), 404 pytest.param( 405 wire( 406 header(qdcount=2), 407 question( 408 root(), 409 dns.rdatatype.RdataType.A, 410 dns.rdataclass.RdataClass.IN, 411 ), 412 question( 413 root(), 414 dns.rdatatype.RdataType.A, 415 dns.rdataclass.RdataClass(2), 416 ), 417 ), 418 formerr_response_header(), 419 id="twoquestionclasses", 420 ), 421 pytest.param( 422 wire( 423 header(arcount=1), 424 a_rr(owner=oversized_name()), 425 ), 426 formerr_response_header(), 427 id="badrecordname", 428 ), 429 pytest.param( 430 wire( 431 header(arcount=2), 432 a_rr(), 433 rr( 434 root(), 435 dns.rdatatype.RdataType(65280), 436 dns.rdataclass.RdataClass(256), 437 rdata=a_rdata(), 438 ), 439 ), 440 formerr_response_header(), 441 id="wrongclass", 442 ), 443 pytest.param( 444 wire( 445 header(qdcount=1, arcount=1), 446 question(root(), dns.rdatatype.RdataType.A), 447 key_rr(rdclass=dns.rdataclass.RdataClass(2)), 448 ), 449 wire( 450 formerr_response_header(qdcount=1), 451 question(root(), dns.rdatatype.RdataType.A), 452 ), 453 id="keyclass", 454 ), 455 pytest.param( 456 wire( 457 header(arcount=1), 458 # OPT owner should be root 459 opt_rr(owner=name("A.")), 460 ), 461 formerr_response_header(), 462 id="optwrongname", 463 ), 464 pytest.param( 465 wire( 466 header(arcount=1), 467 malformed_rrsig_rr(), 468 ), 469 formerr_response_header(), 470 id="malformedrrsig", 471 ), 472 pytest.param( 473 wire( 474 header(opcode=dns.opcode.UPDATE, nscount=1), 475 rr( 476 root(), 477 dns.rdatatype.RdataType.A, 478 dns.rdataclass.RdataClass.ANY, 479 ttl=0, 480 # Non-empty rdata for DELETE type 481 rdata=b"\x00", 482 ), 483 ), 484 formerr_response_header(opcode=dns.opcode.UPDATE), 485 id="malformeddeltype", 486 ), 487 pytest.param( 488 wire( 489 header(arcount=1), 490 # Class should be ANY not IN 491 tsig_rr(rdclass=dns.rdataclass.RdataClass.IN), 492 ), 493 formerr_response_header(), 494 id="tsigwrongclass", 495 ), 496 pytest.param( 497 wire( 498 # Test non-zero message ID is preserved in the response 499 header(message_id=67, arcount=2), 500 tsig_rr(), 501 # TSIG should be the last record 502 a_rr(), 503 ), 504 formerr_response_header(message_id=67), 505 id="tsignotlast", 506 ), 507 ], 508 ) 509 def test_formerr( 510 query_wire: bytes, 511 expected_wire: bytes, 512 named_port: int, 513 ns1, 514 ) -> None: 515 response_wire = query_raw_tcp(ns1.ip, named_port, query_wire) 516 assert response_wire == expected_wire 517
Indexes created Thu Sep 03 00:25:33 UTC 2026