Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/edns.py: 42%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

399 statements  

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license 

2 

3# Copyright (C) 2009-2017 Nominum, Inc. 

4# 

5# Permission to use, copy, modify, and distribute this software and its 

6# documentation for any purpose with or without fee is hereby granted, 

7# provided that the above copyright notice and this permission notice 

8# appear in all copies. 

9# 

10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES 

11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF 

12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR 

13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES 

14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN 

15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT 

16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. 

17 

18"""EDNS Options""" 

19 

20import binascii 

21import math 

22import socket 

23import struct 

24from typing import Any 

25 

26import dns.enum 

27import dns.inet 

28import dns.ipv4 

29import dns.ipv6 

30import dns.name 

31import dns.rdata 

32import dns.wire 

33 

34 

35class OptionType(dns.enum.IntEnum): 

36 """EDNS option type codes""" 

37 

38 #: NSID 

39 NSID = 3 

40 #: DAU 

41 DAU = 5 

42 #: DHU 

43 DHU = 6 

44 #: N3U 

45 N3U = 7 

46 #: ECS (client-subnet) 

47 ECS = 8 

48 #: EXPIRE 

49 EXPIRE = 9 

50 #: COOKIE 

51 COOKIE = 10 

52 #: KEEPALIVE 

53 KEEPALIVE = 11 

54 #: PADDING 

55 PADDING = 12 

56 #: CHAIN 

57 CHAIN = 13 

58 #: EDE (extended-dns-error) 

59 EDE = 15 

60 #: REPORTCHANNEL 

61 REPORTCHANNEL = 18 

62 #: EDE-EXTRA-TEXT-LANGUAGE (Extended DNS Error EXTRA-TEXT language) 

63 EDE_EXTRA_TEXT_LANGUAGE = 22 

64 #: FILTERING-CONTACT 

65 FILTERING_CONTACT = 23 

66 #: FILTERING-ORGANIZATION 

67 FILTERING_ORGANIZATION = 24 

68 #: FILTERING-DB 

69 FILTERING_DB = 25 

70 

71 @classmethod 

72 def _maximum(cls): 

73 return 65535 

74 

75 

76class Option: 

77 """Base class for all EDNS option types.""" 

78 

79 def __init__(self, otype: OptionType | str): 

80 """Initialize an option. 

81 

82 :param otype: The option type. 

83 :type otype: :py:class:`dns.edns.OptionType` 

84 """ 

85 self.otype = OptionType.make(otype) 

86 

87 def to_wire(self, file: Any | None = None) -> bytes | None: 

88 """Convert an option to wire format. 

89 

90 :rtype: bytes or ``None`` 

91 """ 

92 raise NotImplementedError # pragma: no cover 

93 

94 def to_text(self) -> str: 

95 raise NotImplementedError # pragma: no cover 

96 

97 def to_generic(self) -> "GenericOption": 

98 """Create a :py:class:`dns.edns.GenericOption` equivalent of this option. 

99 

100 :rtype: :py:class:`dns.edns.GenericOption` 

101 """ 

102 wire = self.to_wire() 

103 assert wire is not None # for mypy 

104 return GenericOption(self.otype, wire) 

105 

106 @classmethod 

107 def from_wire_parser(cls, otype: OptionType, parser: "dns.wire.Parser") -> "Option": 

108 """Build an EDNS option object from wire format. 

109 

110 :param otype: The option type. 

111 :type otype: :py:class:`dns.edns.OptionType` 

112 :param parser: The parser, restricted to the option length. 

113 :type parser: :py:class:`dns.wire.Parser` 

114 :rtype: :py:class:`dns.edns.Option` 

115 """ 

116 raise NotImplementedError # pragma: no cover 

117 

118 def _cmp(self, other): 

119 """Compare an EDNS option with another option of the same type. 

120 

121 Returns < 0 if < *other*, 0 if == *other*, and > 0 if > *other*. 

122 """ 

123 wire = self.to_wire() 

124 owire = other.to_wire() 

125 if wire == owire: 

126 return 0 

127 if wire > owire: 

128 return 1 

129 return -1 

130 

131 def __eq__(self, other): 

132 if not isinstance(other, Option): 

133 return False 

134 if self.otype != other.otype: 

135 return False 

136 return self._cmp(other) == 0 

137 

138 def __ne__(self, other): 

139 if not isinstance(other, Option): 

140 return True 

141 if self.otype != other.otype: 

142 return True 

143 return self._cmp(other) != 0 

144 

145 def __lt__(self, other): 

146 if not isinstance(other, Option) or self.otype != other.otype: 

147 return NotImplemented 

148 return self._cmp(other) < 0 

149 

150 def __le__(self, other): 

151 if not isinstance(other, Option) or self.otype != other.otype: 

152 return NotImplemented 

153 return self._cmp(other) <= 0 

154 

155 def __ge__(self, other): 

156 if not isinstance(other, Option) or self.otype != other.otype: 

157 return NotImplemented 

158 return self._cmp(other) >= 0 

159 

160 def __gt__(self, other): 

161 if not isinstance(other, Option) or self.otype != other.otype: 

162 return NotImplemented 

163 return self._cmp(other) > 0 

164 

165 def __str__(self): 

166 return self.to_text() 

167 

168 

169class GenericOption(Option): # lgtm[py/missing-equals] 

170 """Generic Option Class 

171 

172 This class is used for EDNS option types for which we have no better 

173 implementation. 

174 """ 

175 

176 def __init__(self, otype: OptionType | str, data: bytes | str): 

177 super().__init__(otype) 

178 self.data = dns.rdata.Rdata._as_bytes(data, True) 

179 

180 def to_wire(self, file: Any | None = None) -> bytes | None: 

181 if file: 

182 file.write(self.data) 

183 return None 

184 else: 

185 return self.data 

186 

187 def to_text(self) -> str: 

188 return f"Generic {self.otype}" 

189 

190 def to_generic(self) -> "GenericOption": 

191 return self 

192 

193 @classmethod 

194 def from_wire_parser( 

195 cls, otype: OptionType | str, parser: "dns.wire.Parser" 

196 ) -> Option: 

197 return cls(otype, parser.get_remaining()) 

198 

199 

200class ECSOption(Option): # lgtm[py/missing-equals] 

201 """EDNS Client Subnet (ECS, RFC7871)""" 

202 

203 def __init__(self, address: str, srclen: int | None = None, scopelen: int = 0): 

204 """Initialize an ECSOption. 

205 

206 :param address: The client address information. 

207 :type address: str 

208 :param srclen: The source prefix length (leftmost number of bits of the 

209 address to be used for the lookup). Defaults to 24 for IPv4 and 56 

210 for IPv6. 

211 :type srclen: int or ``None`` 

212 :param scopelen: The scope prefix length. Must be 0 in queries; should 

213 be set in responses. 

214 :type scopelen: int 

215 """ 

216 

217 super().__init__(OptionType.ECS) 

218 af = dns.inet.af_for_address(address) 

219 

220 if af == socket.AF_INET6: 

221 self.family = 2 

222 if srclen is None: 

223 srclen = 56 

224 address = dns.rdata.Rdata._as_ipv6_address(address) 

225 srclen = dns.rdata.Rdata._as_int(srclen, 0, 128) 

226 scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 128) 

227 elif af == socket.AF_INET: 

228 self.family = 1 

229 if srclen is None: 

230 srclen = 24 

231 address = dns.rdata.Rdata._as_ipv4_address(address) 

232 srclen = dns.rdata.Rdata._as_int(srclen, 0, 32) 

233 scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 32) 

234 else: # pragma: no cover (this will never happen) 

235 raise ValueError("Bad address family") 

236 

237 assert srclen is not None 

238 self.address = address 

239 self.srclen = srclen 

240 self.scopelen = scopelen 

241 

242 addrdata = dns.inet.inet_pton(af, address) 

243 nbytes = int(math.ceil(srclen / 8.0)) 

244 

245 # Truncate to srclen and pad to the end of the last octet needed 

246 # See RFC section 6 

247 self.addrdata = addrdata[:nbytes] 

248 nbits = srclen % 8 

249 if nbits != 0: 

250 last = struct.pack("B", ord(self.addrdata[-1:]) & (0xFF << (8 - nbits))) 

251 self.addrdata = self.addrdata[:-1] + last 

252 

253 def to_text(self) -> str: 

254 return f"ECS {self.address}/{self.srclen} scope/{self.scopelen}" 

255 

256 @staticmethod 

257 def from_text(text: str) -> Option: 

258 """Convert a string into a `dns.edns.ECSOption` 

259 

260 *text*, a `str`, the text form of the option. 

261 

262 Returns a `dns.edns.ECSOption`. 

263 

264 Examples: 

265 

266 >>> import dns.edns 

267 >>> 

268 >>> # basic example 

269 >>> dns.edns.ECSOption.from_text('1.2.3.4/24') 

270 >>> 

271 >>> # also understands scope 

272 >>> dns.edns.ECSOption.from_text('1.2.3.4/24/32') 

273 >>> 

274 >>> # IPv6 

275 >>> dns.edns.ECSOption.from_text('2001:4b98::1/64/64') 

276 >>> 

277 >>> # it understands results from `dns.edns.ECSOption.to_text()` 

278 >>> dns.edns.ECSOption.from_text('ECS 1.2.3.4/24 scope/32') 

279 """ 

280 optional_prefix = "ECS" 

281 scope_prefix = "scope/" 

282 tokens = text.split() 

283 ecs_text = None 

284 if len(tokens) == 1: 

285 ecs_text = tokens[0] 

286 elif len(tokens) == 2: 

287 if tokens[0] != optional_prefix: 

288 raise ValueError(f'could not parse ECS from "{text}"') 

289 ecs_text = tokens[1] 

290 elif ( 

291 len(tokens) == 3 

292 and tokens[0] == optional_prefix 

293 and tokens[1].count("/") == 1 

294 and tokens[2].startswith(scope_prefix) 

295 ): 

296 # The form emitted by to_text(). 

297 ecs_text = f"{tokens[1]}/{tokens[2][len(scope_prefix) :]}" 

298 else: 

299 raise ValueError(f'could not parse ECS from "{text}"') 

300 n_slashes = ecs_text.count("/") 

301 if n_slashes == 1: 

302 address, tsrclen = ecs_text.split("/") 

303 tscope = "0" 

304 elif n_slashes == 2: 

305 address, tsrclen, tscope = ecs_text.split("/") 

306 else: 

307 raise ValueError(f'could not parse ECS from "{text}"') 

308 try: 

309 scope = int(tscope) 

310 except ValueError: 

311 raise ValueError("invalid scope " + f'"{tscope}": scope must be an integer') 

312 try: 

313 srclen = int(tsrclen) 

314 except ValueError: 

315 raise ValueError( 

316 "invalid srclen " + f'"{tsrclen}": srclen must be an integer' 

317 ) 

318 return ECSOption(address, srclen, scope) 

319 

320 def to_wire(self, file: Any | None = None) -> bytes | None: 

321 value = ( 

322 struct.pack("!HBB", self.family, self.srclen, self.scopelen) + self.addrdata 

323 ) 

324 if file: 

325 file.write(value) 

326 return None 

327 else: 

328 return value 

329 

330 @classmethod 

331 def from_wire_parser( 

332 cls, otype: OptionType | str, parser: "dns.wire.Parser" 

333 ) -> Option: 

334 family, src, scope = parser.get_struct("!HBB") 

335 if family == 1: 

336 bits = 32 

337 elif family == 2: 

338 bits = 128 

339 else: 

340 raise ValueError("unsupported family") 

341 # RFC 7871 Section 6: the source and scope prefix lengths are the 

342 # significant bits of ADDRESS and cannot exceed the width of the 

343 # address family. 

344 if src > bits or scope > bits: 

345 raise ValueError("invalid ECS prefix length") 

346 addrlen = int(math.ceil(src / 8.0)) 

347 prefix = parser.get_bytes(addrlen) 

348 # RFC 7871 Sections 6 and 7.4: bits of ADDRESS beyond the source 

349 # prefix length are padding and must be zero. Reject a non-zero pad 

350 # rather than letting the ECSOption constructor mask it off, which 

351 # would make from_wire and to_wire disagree on the same option. 

352 rest = src % 8 

353 if rest != 0 and prefix[-1] & (0xFF >> rest): 

354 raise ValueError("non-zero ECS address padding bits") 

355 if family == 1: 

356 addr = dns.ipv4.inet_ntoa(prefix + b"\x00" * (4 - addrlen)) 

357 else: 

358 addr = dns.ipv6.inet_ntoa(prefix + b"\x00" * (16 - addrlen)) 

359 return cls(addr, src, scope) 

360 

361 

362class EDECode(dns.enum.IntEnum): 

363 """Extended DNS Error (EDE) codes""" 

364 

365 OTHER = 0 

366 UNSUPPORTED_DNSKEY_ALGORITHM = 1 

367 UNSUPPORTED_DS_DIGEST_TYPE = 2 

368 STALE_ANSWER = 3 

369 FORGED_ANSWER = 4 

370 DNSSEC_INDETERMINATE = 5 

371 DNSSEC_BOGUS = 6 

372 SIGNATURE_EXPIRED = 7 

373 SIGNATURE_NOT_YET_VALID = 8 

374 DNSKEY_MISSING = 9 

375 RRSIGS_MISSING = 10 

376 NO_ZONE_KEY_BIT_SET = 11 

377 NSEC_MISSING = 12 

378 CACHED_ERROR = 13 

379 NOT_READY = 14 

380 BLOCKED = 15 

381 CENSORED = 16 

382 FILTERED = 17 

383 PROHIBITED = 18 

384 STALE_NXDOMAIN_ANSWER = 19 

385 NOT_AUTHORITATIVE = 20 

386 NOT_SUPPORTED = 21 

387 NO_REACHABLE_AUTHORITY = 22 

388 NETWORK_ERROR = 23 

389 INVALID_DATA = 24 

390 SIGNATURE_EXPIRED_BEFORE_VALID = 25 

391 TOO_EARLY = 26 

392 UNSUPPORTED_NSEC3_ITERATIONS_VALUE = 27 

393 UNABLE_TO_CONFORM_TO_POLICY = 28 

394 SYNTHESIZED = 29 

395 INVALID_QUERY_TYPE = 30 

396 RATE_LIMITED = 31 

397 OVER_QUOTA = 32 

398 

399 @classmethod 

400 def _maximum(cls): 

401 return 65535 

402 

403 

404class EDEOption(Option): # lgtm[py/missing-equals] 

405 """Extended DNS Error (EDE, RFC8914)""" 

406 

407 _preserve_case = {"DNSKEY", "DS", "DNSSEC", "RRSIGs", "NSEC", "NXDOMAIN"} 

408 

409 def __init__(self, code: EDECode | str, text: str | None = None): 

410 """Initialize an EDEOption. 

411 

412 :param code: The info code of the extended error. 

413 :type code: :py:class:`dns.edns.EDECode` or str 

414 :param text: Additional information about the error. 

415 :type text: str or ``None`` 

416 """ 

417 

418 super().__init__(OptionType.EDE) 

419 

420 self.code = EDECode.make(code) 

421 if text is not None and not isinstance(text, str): 

422 raise ValueError("text must be string or None") 

423 self.text = text 

424 

425 def to_text(self) -> str: 

426 output = f"EDE {self.code}" 

427 if self.code in EDECode: 

428 desc = EDECode.to_text(self.code) 

429 desc = " ".join( 

430 word if word in self._preserve_case else word.title() 

431 for word in desc.split("_") 

432 ) 

433 output += f" ({desc})" 

434 if self.text is not None: 

435 output += f": {self.text}" 

436 return output 

437 

438 def to_wire(self, file: Any | None = None) -> bytes | None: 

439 value = struct.pack("!H", self.code) 

440 if self.text is not None: 

441 value += self.text.encode("utf8") 

442 

443 if file: 

444 file.write(value) 

445 return None 

446 else: 

447 return value 

448 

449 @classmethod 

450 def from_wire_parser( 

451 cls, otype: OptionType | str, parser: "dns.wire.Parser" 

452 ) -> Option: 

453 code = EDECode.make(parser.get_uint16()) 

454 text = parser.get_remaining() 

455 

456 if text: 

457 if text[-1] == 0: # text MAY be null-terminated 

458 text = text[:-1] 

459 btext = text.decode("utf8") 

460 else: 

461 btext = None 

462 

463 return cls(code, btext) 

464 

465 

466class NSIDOption(Option): 

467 def __init__(self, nsid: bytes): 

468 super().__init__(OptionType.NSID) 

469 self.nsid = nsid 

470 

471 def to_wire(self, file: Any = None) -> bytes | None: 

472 if file: 

473 file.write(self.nsid) 

474 return None 

475 else: 

476 return self.nsid 

477 

478 def to_text(self) -> str: 

479 if all(c >= 0x20 and c <= 0x7E for c in self.nsid): 

480 # All ASCII printable, so it's probably a string. 

481 value = self.nsid.decode() 

482 else: 

483 value = binascii.hexlify(self.nsid).decode() 

484 return f"NSID {value}" 

485 

486 @classmethod 

487 def from_wire_parser( 

488 cls, otype: OptionType | str, parser: dns.wire.Parser 

489 ) -> Option: 

490 return cls(parser.get_remaining()) 

491 

492 

493class CookieOption(Option): 

494 def __init__(self, client: bytes, server: bytes): 

495 super().__init__(OptionType.COOKIE) 

496 self.client = client 

497 self.server = server 

498 if len(client) != 8: 

499 raise ValueError("client cookie must be 8 bytes") 

500 if len(server) != 0 and (len(server) < 8 or len(server) > 32): 

501 raise ValueError("server cookie must be empty or between 8 and 32 bytes") 

502 

503 def to_wire(self, file: Any = None) -> bytes | None: 

504 if file: 

505 file.write(self.client) 

506 if len(self.server) > 0: 

507 file.write(self.server) 

508 return None 

509 else: 

510 return self.client + self.server 

511 

512 def to_text(self) -> str: 

513 client = binascii.hexlify(self.client).decode() 

514 if len(self.server) > 0: 

515 server = binascii.hexlify(self.server).decode() 

516 else: 

517 server = "" 

518 return f"COOKIE {client}{server}" 

519 

520 @classmethod 

521 def from_wire_parser( 

522 cls, otype: OptionType | str, parser: dns.wire.Parser 

523 ) -> Option: 

524 return cls(parser.get_bytes(8), parser.get_remaining()) 

525 

526 

527class ReportChannelOption(Option): 

528 # RFC 9567 

529 def __init__(self, agent_domain: dns.name.Name): 

530 super().__init__(OptionType.REPORTCHANNEL) 

531 self.agent_domain = agent_domain 

532 

533 def to_wire(self, file: Any = None) -> bytes | None: 

534 return self.agent_domain.to_wire(file) 

535 

536 def to_text(self) -> str: 

537 return "REPORTCHANNEL " + self.agent_domain.to_text() 

538 

539 @classmethod 

540 def from_wire_parser( 

541 cls, otype: OptionType | str, parser: dns.wire.Parser 

542 ) -> Option: 

543 return cls(parser.get_name()) 

544 

545 

546class EDEExtraTextLanguageOption(Option): 

547 """Extended DNS Error EXTRA-TEXT language (EDE-EXTRA-TEXT-LANGUAGE)""" 

548 

549 def __init__(self, language: str): 

550 """Initialize an EDEExtraTextLanguageOption. 

551 

552 :param language: The language of EXTRA-TEXT in the EDE option. 

553 :type language: str 

554 """ 

555 

556 super().__init__(OptionType.EDE_EXTRA_TEXT_LANGUAGE) 

557 self.language = language 

558 

559 def to_wire(self, file: Any = None) -> bytes | None: 

560 if file: 

561 file.write(self.language.encode("utf8")) 

562 return None 

563 else: 

564 return self.language.encode() 

565 

566 def to_text(self) -> str: 

567 return f"EDE-EXTRA-TEXT-LANGUAGE {self.language}" 

568 

569 @classmethod 

570 def from_wire_parser( 

571 cls, otype: OptionType | str, parser: dns.wire.Parser 

572 ) -> Option: 

573 return cls(parser.get_remaining().decode("utf8")) 

574 

575 

576class FilteringContactOption(Option): 

577 """Filtering contact (FILTERING-CONTACT)""" 

578 

579 def __init__(self, contact: str): 

580 """Initialize a FilteringContactOption. 

581 

582 :param contact: A filtering contact URI as a string. 

583 :type contact: str 

584 """ 

585 

586 super().__init__(OptionType.FILTERING_CONTACT) 

587 self.contact = contact 

588 

589 def to_wire(self, file: Any = None) -> bytes | None: 

590 if file: 

591 file.write(self.contact.encode("utf8")) 

592 return None 

593 else: 

594 return self.contact.encode() 

595 

596 def to_text(self) -> str: 

597 return f"FILTERING-CONTACT {self.contact}" 

598 

599 @classmethod 

600 def from_wire_parser( 

601 cls, otype: OptionType | str, parser: dns.wire.Parser 

602 ) -> Option: 

603 return cls(parser.get_remaining().decode("utf8")) 

604 

605 

606class FilteringOrganizationOption(Option): 

607 """Filtering organization (FILTERING-ORGANIZATION)""" 

608 

609 def __init__(self, organization: str): 

610 """Initialize a FilteringOrganizationOption. 

611 

612 :param organization: The filtering organization. 

613 :type organization: str 

614 """ 

615 

616 super().__init__(OptionType.FILTERING_ORGANIZATION) 

617 self.organization = organization 

618 

619 def to_wire(self, file: Any = None) -> bytes | None: 

620 if file: 

621 file.write(self.organization.encode("utf8")) 

622 return None 

623 else: 

624 return self.organization.encode() 

625 

626 def to_text(self) -> str: 

627 return f"FILTERING-ORGANIZATION {self.organization}" 

628 

629 @classmethod 

630 def from_wire_parser( 

631 cls, otype: OptionType | str, parser: dns.wire.Parser 

632 ) -> Option: 

633 return cls(parser.get_remaining().decode("utf8")) 

634 

635 

636class FilteringDBOption(Option): 

637 """Filtering DB (FILTERING-DB)""" 

638 

639 def __init__(self, db: str): 

640 """Initialize a FilteringDBOption. 

641 

642 :param db: The filtering database containing the identifier, 

643 name, or description of the filtering database 

644 against which a matched query caused the filtering to 

645 occur. 

646 :type db: str 

647 

648 """ 

649 

650 super().__init__(OptionType.FILTERING_DB) 

651 self.db = db 

652 

653 def to_wire(self, file: Any = None) -> bytes | None: 

654 if file: 

655 file.write(self.db.encode("utf8")) 

656 return None 

657 else: 

658 return self.db.encode() 

659 

660 def to_text(self) -> str: 

661 return f"FILTERING-DB {self.db}" 

662 

663 @classmethod 

664 def from_wire_parser( 

665 cls, otype: OptionType | str, parser: dns.wire.Parser 

666 ) -> Option: 

667 return cls(parser.get_remaining().decode("utf8")) 

668 

669 

670_type_to_class: dict[OptionType, Any] = { 

671 OptionType.ECS: ECSOption, 

672 OptionType.EDE: EDEOption, 

673 OptionType.NSID: NSIDOption, 

674 OptionType.COOKIE: CookieOption, 

675 OptionType.REPORTCHANNEL: ReportChannelOption, 

676 OptionType.EDE_EXTRA_TEXT_LANGUAGE: EDEExtraTextLanguageOption, 

677 OptionType.FILTERING_CONTACT: FilteringContactOption, 

678 OptionType.FILTERING_ORGANIZATION: FilteringOrganizationOption, 

679 OptionType.FILTERING_DB: FilteringDBOption, 

680} 

681 

682 

683def get_option_class(otype: OptionType) -> Any: 

684 """Return the class for the specified option type. 

685 

686 The GenericOption class is used if a more specific class is not 

687 known. 

688 """ 

689 

690 cls = _type_to_class.get(otype) 

691 if cls is None: 

692 cls = GenericOption 

693 return cls 

694 

695 

696def option_from_wire_parser( 

697 otype: OptionType | str, parser: "dns.wire.Parser" 

698) -> Option: 

699 """Build an EDNS option object from wire format. 

700 

701 :param otype: The option type. 

702 :type otype: int or :py:class:`dns.edns.OptionType` 

703 :param parser: The parser, restricted to the option length. 

704 :type parser: :py:class:`dns.wire.Parser` 

705 :rtype: :py:class:`dns.edns.Option` 

706 """ 

707 otype = OptionType.make(otype) 

708 cls = get_option_class(otype) 

709 return cls.from_wire_parser(otype, parser) 

710 

711 

712def option_from_wire( 

713 otype: OptionType | str, wire: bytes, current: int, olen: int 

714) -> Option: 

715 """Build an EDNS option object from wire format. 

716 

717 :param otype: The option type. 

718 :type otype: int or :py:class:`dns.edns.OptionType` 

719 :param wire: The wire-format message. 

720 :type wire: bytes 

721 :param current: The offset in *wire* of the beginning of the rdata. 

722 :type current: int 

723 :param olen: The length of the wire-format option data. 

724 :type olen: int 

725 :rtype: :py:class:`dns.edns.Option` 

726 """ 

727 parser = dns.wire.Parser(wire, current) 

728 with parser.restrict_to(olen): 

729 return option_from_wire_parser(otype, parser) 

730 

731 

732def register_type(implementation: Any, otype: OptionType) -> None: 

733 """Register the implementation of an option type. 

734 

735 :param implementation: A subclass of :py:class:`dns.edns.Option`. 

736 :param otype: The option type. 

737 :type otype: :py:class:`dns.edns.OptionType` 

738 """ 

739 

740 _type_to_class[otype] = implementation 

741 

742 

743### BEGIN generated OptionType constants 

744 

745NSID = OptionType.NSID 

746DAU = OptionType.DAU 

747DHU = OptionType.DHU 

748N3U = OptionType.N3U 

749ECS = OptionType.ECS 

750EXPIRE = OptionType.EXPIRE 

751COOKIE = OptionType.COOKIE 

752KEEPALIVE = OptionType.KEEPALIVE 

753PADDING = OptionType.PADDING 

754CHAIN = OptionType.CHAIN 

755EDE = OptionType.EDE 

756REPORTCHANNEL = OptionType.REPORTCHANNEL 

757EDE_EXTRA_TEXT_LANGUAGE = OptionType.EDE_EXTRA_TEXT_LANGUAGE 

758FILTERING_CONTACT = OptionType.FILTERING_CONTACT 

759FILTERING_ORGANIZATION = OptionType.FILTERING_ORGANIZATION 

760FILTERING_DB = OptionType.FILTERING_DB 

761 

762### END generated OptionType constants