codekingpro/portable-devtools
114k
1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3# Copyright (C) 2001-2017 Nominum, Inc.4#5# Permission to use, copy, modify, and distribute this software and its6# documentation for any purpose with or without fee is hereby granted,7# provided that the above copyright notice and this permission notice8# appear in all copies.9#10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.17 18"""DNS Messages"""19 20import contextlib21import io22import time23from typing import Any, Dict, List, Optional, Tuple, Union24 25import dns.edns26import dns.entropy27import dns.enum28import dns.exception29import dns.flags30import dns.name31import dns.opcode32import dns.rcode33import dns.rdata34import dns.rdataclass35import dns.rdatatype36import dns.rdtypes.ANY.OPT37import dns.rdtypes.ANY.TSIG38import dns.renderer39import dns.rrset40import dns.tsig41import dns.ttl42import dns.wire43 44 45class ShortHeader(dns.exception.FormError):46 """The DNS packet passed to from_wire() is too short."""47 48 49class TrailingJunk(dns.exception.FormError):50 """The DNS packet passed to from_wire() has extra junk at the end of it."""51 52 53class UnknownHeaderField(dns.exception.DNSException):54 """The header field name was not recognized when converting from text55 into a message."""56 57 58class BadEDNS(dns.exception.FormError):59 """An OPT record occurred somewhere other than60 the additional data section."""61 62 63class BadTSIG(dns.exception.FormError):64 """A TSIG record occurred somewhere other than the end of65 the additional data section."""66 67 68class UnknownTSIGKey(dns.exception.DNSException):69 """A TSIG with an unknown key was received."""70 71 72class Truncated(dns.exception.DNSException):73 """The truncated flag is set."""74 75 supp_kwargs = {"message"}76 77 # We do this as otherwise mypy complains about unexpected keyword argument78 # idna_exception79 def __init__(self, *args, **kwargs):80 super().__init__(*args, **kwargs)81 82 def message(self):83 """As much of the message as could be processed.84 85 Returns a ``dns.message.Message``.86 """87 return self.kwargs["message"]88 89 90class NotQueryResponse(dns.exception.DNSException):91 """Message is not a response to a query."""92 93 94class ChainTooLong(dns.exception.DNSException):95 """The CNAME chain is too long."""96 97 98class AnswerForNXDOMAIN(dns.exception.DNSException):99 """The rcode is NXDOMAIN but an answer was found."""100 101 102class NoPreviousName(dns.exception.SyntaxError):103 """No previous name was known."""104 105 106class MessageSection(dns.enum.IntEnum):107 """Message sections"""108 109 QUESTION = 0110 ANSWER = 1111 AUTHORITY = 2112 ADDITIONAL = 3113 114 @classmethod115 def _maximum(cls):116 return 3117 118 119class MessageError:120 def __init__(self, exception: Exception, offset: int):121 self.exception = exception122 self.offset = offset123 124 125DEFAULT_EDNS_PAYLOAD = 1232126MAX_CHAIN = 16127 128IndexKeyType = Tuple[129 int,130 dns.name.Name,131 dns.rdataclass.RdataClass,132 dns.rdatatype.RdataType,133 Optional[dns.rdatatype.RdataType],134 Optional[dns.rdataclass.RdataClass],135]136IndexType = Dict[IndexKeyType, dns.rrset.RRset]137SectionType = Union[int, str, List[dns.rrset.RRset]]138 139 140class Message:141 """A DNS message."""142 143 _section_enum = MessageSection144 145 def __init__(self, id: Optional[int] = None):146 if id is None:147 self.id = dns.entropy.random_16()148 else:149 self.id = id150 self.flags = 0151 self.sections: List[List[dns.rrset.RRset]] = [[], [], [], []]152 self.opt: Optional[dns.rrset.RRset] = None153 self.request_payload = 0154 self.pad = 0155 self.keyring: Any = None156 self.tsig: Optional[dns.rrset.RRset] = None157 self.request_mac = b""158 self.xfr = False159 self.origin: Optional[dns.name.Name] = None160 self.tsig_ctx: Optional[Any] = None161 self.index: IndexType = {}162 self.errors: List[MessageError] = []163 self.time = 0.0164 165 @property166 def question(self) -> List[dns.rrset.RRset]:167 """The question section."""168 return self.sections[0]169 170 @question.setter171 def question(self, v):172 self.sections[0] = v173 174 @property175 def answer(self) -> List[dns.rrset.RRset]:176 """The answer section."""177 return self.sections[1]178 179 @answer.setter180 def answer(self, v):181 self.sections[1] = v182 183 @property184 def authority(self) -> List[dns.rrset.RRset]:185 """The authority section."""186 return self.sections[2]187 188 @authority.setter189 def authority(self, v):190 self.sections[2] = v191 192 @property193 def additional(self) -> List[dns.rrset.RRset]:194 """The additional data section."""195 return self.sections[3]196 197 @additional.setter198 def additional(self, v):199 self.sections[3] = v200 201 def __repr__(self):202 return "<DNS message, ID " + repr(self.id) + ">"203 204 def __str__(self):205 return self.to_text()206 207 def to_text(208 self,209 origin: Optional[dns.name.Name] = None,210 relativize: bool = True,211 **kw: Dict[str, Any],212 ) -> str:213 """Convert the message to text.214 215 The *origin*, *relativize*, and any other keyword216 arguments are passed to the RRset ``to_wire()`` method.217 218 Returns a ``str``.219 """220 221 s = io.StringIO()222 s.write("id %d\n" % self.id)223 s.write("opcode %s\n" % dns.opcode.to_text(self.opcode()))224 s.write("rcode %s\n" % dns.rcode.to_text(self.rcode()))225 s.write("flags %s\n" % dns.flags.to_text(self.flags))226 if self.edns >= 0:227 s.write("edns %s\n" % self.edns)228 if self.ednsflags != 0:229 s.write("eflags %s\n" % dns.flags.edns_to_text(self.ednsflags))230 s.write("payload %d\n" % self.payload)231 for opt in self.options:232 s.write("option %s\n" % opt.to_text())233 for name, which in self._section_enum.__members__.items():234 s.write(f";{name}\n")235 for rrset in self.section_from_number(which):236 s.write(rrset.to_text(origin, relativize, **kw))237 s.write("\n")238 #239 # We strip off the final \n so the caller can print the result without240 # doing weird things to get around eccentricities in Python print241 # formatting242 #243 return s.getvalue()[:-1]244 245 def __eq__(self, other):246 """Two messages are equal if they have the same content in the247 header, question, answer, and authority sections.248 249 Returns a ``bool``.250 """251 252 if not isinstance(other, Message):253 return False254 if self.id != other.id:255 return False256 if self.flags != other.flags:257 return False258 for i, section in enumerate(self.sections):259 other_section = other.sections[i]260 for n in section:261 if n not in other_section:262 return False263 for n in other_section:264 if n not in section:265 return False266 return True267 268 def __ne__(self, other):269 return not self.__eq__(other)270 271 def is_response(self, other: "Message") -> bool:272 """Is *other*, also a ``dns.message.Message``, a response to this273 message?274 275 Returns a ``bool``.276 """277 278 if (279 other.flags & dns.flags.QR == 0280 or self.id != other.id281 or dns.opcode.from_flags(self.flags) != dns.opcode.from_flags(other.flags)282 ):283 return False284 if other.rcode() in {285 dns.rcode.FORMERR,286 dns.rcode.SERVFAIL,287 dns.rcode.NOTIMP,288 dns.rcode.REFUSED,289 }:290 # We don't check the question section in these cases if291 # the other question section is empty, even though they292 # still really ought to have a question section.293 if len(other.question) == 0:294 return True295 if dns.opcode.is_update(self.flags):296 # This is assuming the "sender doesn't include anything297 # from the update", but we don't care to check the other298 # case, which is that all the sections are returned and299 # identical.300 return True301 for n in self.question:302 if n not in other.question:303 return False304 for n in other.question:305 if n not in self.question:306 return False307 return True308 309 def section_number(self, section: List[dns.rrset.RRset]) -> int:310 """Return the "section number" of the specified section for use311 in indexing.312 313 *section* is one of the section attributes of this message.314 315 Raises ``ValueError`` if the section isn't known.316 317 Returns an ``int``.318 """319 320 for i, our_section in enumerate(self.sections):321 if section is our_section:322 return self._section_enum(i)323 raise ValueError("unknown section")324 325 def section_from_number(self, number: int) -> List[dns.rrset.RRset]:326 """Return the section list associated with the specified section327 number.328 329 *number* is a section number `int` or the text form of a section330 name.331 332 Raises ``ValueError`` if the section isn't known.333 334 Returns a ``list``.335 """336 337 section = self._section_enum.make(number)338 return self.sections[section]339 340 def find_rrset(341 self,342 section: SectionType,343 name: dns.name.Name,344 rdclass: dns.rdataclass.RdataClass,345 rdtype: dns.rdatatype.RdataType,346 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,347 deleting: Optional[dns.rdataclass.RdataClass] = None,348 create: bool = False,349 force_unique: bool = False,350 idna_codec: Optional[dns.name.IDNACodec] = None,351 ) -> dns.rrset.RRset:352 """Find the RRset with the given attributes in the specified section.353 354 *section*, an ``int`` section number, a ``str`` section name, or one of355 the section attributes of this message. This specifies the356 the section of the message to search. For example::357 358 my_message.find_rrset(my_message.answer, name, rdclass, rdtype)359 my_message.find_rrset(dns.message.ANSWER, name, rdclass, rdtype)360 my_message.find_rrset("ANSWER", name, rdclass, rdtype)361 362 *name*, a ``dns.name.Name`` or ``str``, the name of the RRset.363 364 *rdclass*, an ``int`` or ``str``, the class of the RRset.365 366 *rdtype*, an ``int`` or ``str``, the type of the RRset.367 368 *covers*, an ``int`` or ``str``, the covers value of the RRset.369 The default is ``dns.rdatatype.NONE``.370 371 *deleting*, an ``int``, ``str``, or ``None``, the deleting value of the372 RRset. The default is ``None``.373 374 *create*, a ``bool``. If ``True``, create the RRset if it is not found.375 The created RRset is appended to *section*.376 377 *force_unique*, a ``bool``. If ``True`` and *create* is also ``True``,378 create a new RRset regardless of whether a matching RRset exists379 already. The default is ``False``. This is useful when creating380 DDNS Update messages, as order matters for them.381 382 *idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA383 encoder/decoder. If ``None``, the default IDNA 2003 encoder/decoder384 is used.385 386 Raises ``KeyError`` if the RRset was not found and create was387 ``False``.388 389 Returns a ``dns.rrset.RRset object``.390 """391 392 if isinstance(section, int):393 section_number = section394 section = self.section_from_number(section_number)395 elif isinstance(section, str):396 section_number = self._section_enum.from_text(section)397 section = self.section_from_number(section_number)398 else:399 section_number = self.section_number(section)400 if isinstance(name, str):401 name = dns.name.from_text(name, idna_codec=idna_codec)402 rdtype = dns.rdatatype.RdataType.make(rdtype)403 rdclass = dns.rdataclass.RdataClass.make(rdclass)404 covers = dns.rdatatype.RdataType.make(covers)405 if deleting is not None:406 deleting = dns.rdataclass.RdataClass.make(deleting)407 key = (section_number, name, rdclass, rdtype, covers, deleting)408 if not force_unique:409 if self.index is not None:410 rrset = self.index.get(key)411 if rrset is not None:412 return rrset413 else:414 for rrset in section:415 if rrset.full_match(name, rdclass, rdtype, covers, deleting):416 return rrset417 if not create:418 raise KeyError419 rrset = dns.rrset.RRset(name, rdclass, rdtype, covers, deleting)420 section.append(rrset)421 if self.index is not None:422 self.index[key] = rrset423 return rrset424 425 def get_rrset(426 self,427 section: SectionType,428 name: dns.name.Name,429 rdclass: dns.rdataclass.RdataClass,430 rdtype: dns.rdatatype.RdataType,431 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,432 deleting: Optional[dns.rdataclass.RdataClass] = None,433 create: bool = False,434 force_unique: bool = False,435 idna_codec: Optional[dns.name.IDNACodec] = None,436 ) -> Optional[dns.rrset.RRset]:437 """Get the RRset with the given attributes in the specified section.438 439 If the RRset is not found, None is returned.440 441 *section*, an ``int`` section number, a ``str`` section name, or one of442 the section attributes of this message. This specifies the443 the section of the message to search. For example::444 445 my_message.get_rrset(my_message.answer, name, rdclass, rdtype)446 my_message.get_rrset(dns.message.ANSWER, name, rdclass, rdtype)447 my_message.get_rrset("ANSWER", name, rdclass, rdtype)448 449 *name*, a ``dns.name.Name`` or ``str``, the name of the RRset.450 451 *rdclass*, an ``int`` or ``str``, the class of the RRset.452 453 *rdtype*, an ``int`` or ``str``, the type of the RRset.454 455 *covers*, an ``int`` or ``str``, the covers value of the RRset.456 The default is ``dns.rdatatype.NONE``.457 458 *deleting*, an ``int``, ``str``, or ``None``, the deleting value of the459 RRset. The default is ``None``.460 461 *create*, a ``bool``. If ``True``, create the RRset if it is not found.462 The created RRset is appended to *section*.463 464 *force_unique*, a ``bool``. If ``True`` and *create* is also ``True``,465 create a new RRset regardless of whether a matching RRset exists466 already. The default is ``False``. This is useful when creating467 DDNS Update messages, as order matters for them.468 469 *idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA470 encoder/decoder. If ``None``, the default IDNA 2003 encoder/decoder471 is used.472 473 Returns a ``dns.rrset.RRset object`` or ``None``.474 """475 476 try:477 rrset = self.find_rrset(478 section,479 name,480 rdclass,481 rdtype,482 covers,483 deleting,484 create,485 force_unique,486 idna_codec,487 )488 except KeyError:489 rrset = None490 return rrset491 492 def section_count(self, section: SectionType) -> int:493 """Returns the number of records in the specified section.494 495 *section*, an ``int`` section number, a ``str`` section name, or one of496 the section attributes of this message. This specifies the497 the section of the message to count. For example::498 499 my_message.section_count(my_message.answer)500 my_message.section_count(dns.message.ANSWER)501 my_message.section_count("ANSWER")502 """503 504 if isinstance(section, int):505 section_number = section506 section = self.section_from_number(section_number)507 elif isinstance(section, str):508 section_number = self._section_enum.from_text(section)509 section = self.section_from_number(section_number)510 else:511 section_number = self.section_number(section)512 count = sum(max(1, len(rrs)) for rrs in section)513 if section_number == MessageSection.ADDITIONAL:514 if self.opt is not None:515 count += 1516 if self.tsig is not None:517 count += 1518 return count519 520 def _compute_opt_reserve(self) -> int:521 """Compute the size required for the OPT RR, padding excluded"""522 if not self.opt:523 return 0524 # 1 byte for the root name, 10 for the standard RR fields525 size = 11526 # This would be more efficient if options had a size() method, but we won't527 # worry about that for now. We also don't worry if there is an existing padding528 # option, as it is unlikely and probably harmless, as the worst case is that we529 # may add another, and this seems to be legal.530 for option in self.opt[0].options:531 wire = option.to_wire()532 # We add 4 here to account for the option type and length533 size += len(wire) + 4534 if self.pad:535 # Padding will be added, so again add the option type and length.536 size += 4537 return size538 539 def _compute_tsig_reserve(self) -> int:540 """Compute the size required for the TSIG RR"""541 # This would be more efficient if TSIGs had a size method, but we won't542 # worry about for now. Also, we can't really cope with the potential543 # compressibility of the TSIG owner name, so we estimate with the uncompressed544 # size. We will disable compression when TSIG and padding are both is active545 # so that the padding comes out right.546 if not self.tsig:547 return 0548 f = io.BytesIO()549 self.tsig.to_wire(f)550 return len(f.getvalue())551 552 def to_wire(553 self,554 origin: Optional[dns.name.Name] = None,555 max_size: int = 0,556 multi: bool = False,557 tsig_ctx: Optional[Any] = None,558 prepend_length: bool = False,559 prefer_truncation: bool = False,560 **kw: Dict[str, Any],561 ) -> bytes:562 """Return a string containing the message in DNS compressed wire563 format.564 565 Additional keyword arguments are passed to the RRset ``to_wire()``566 method.567 568 *origin*, a ``dns.name.Name`` or ``None``, the origin to be appended569 to any relative names. If ``None``, and the message has an origin570 attribute that is not ``None``, then it will be used.571 572 *max_size*, an ``int``, the maximum size of the wire format573 output; default is 0, which means "the message's request574 payload, if nonzero, or 65535".575 576 *multi*, a ``bool``, should be set to ``True`` if this message is577 part of a multiple message sequence.578 579 *tsig_ctx*, a ``dns.tsig.HMACTSig`` or ``dns.tsig.GSSTSig`` object, the580 ongoing TSIG context, used when signing zone transfers.581 582 *prepend_length*, a ``bool``, should be set to ``True`` if the caller583 wants the message length prepended to the message itself. This is584 useful for messages sent over TCP, TLS (DoT), or QUIC (DoQ).585 586 *prefer_truncation*, a ``bool``, should be set to ``True`` if the caller587 wants the message to be truncated if it would otherwise exceed the588 maximum length. If the truncation occurs before the additional section,589 the TC bit will be set.590 591 Raises ``dns.exception.TooBig`` if *max_size* was exceeded.592 593 Returns a ``bytes``.594 """595 596 if origin is None and self.origin is not None:597 origin = self.origin598 if max_size == 0:599 if self.request_payload != 0:600 max_size = self.request_payload601 else:602 max_size = 65535603 if max_size < 512:604 max_size = 512605 elif max_size > 65535:606 max_size = 65535607 r = dns.renderer.Renderer(self.id, self.flags, max_size, origin)608 opt_reserve = self._compute_opt_reserve()609 r.reserve(opt_reserve)610 tsig_reserve = self._compute_tsig_reserve()611 r.reserve(tsig_reserve)612 try:613 for rrset in self.question:614 r.add_question(rrset.name, rrset.rdtype, rrset.rdclass)615 for rrset in self.answer:616 r.add_rrset(dns.renderer.ANSWER, rrset, **kw)617 for rrset in self.authority:618 r.add_rrset(dns.renderer.AUTHORITY, rrset, **kw)619 for rrset in self.additional:620 r.add_rrset(dns.renderer.ADDITIONAL, rrset, **kw)621 except dns.exception.TooBig:622 if prefer_truncation:623 if r.section < dns.renderer.ADDITIONAL:624 r.flags |= dns.flags.TC625 else:626 raise627 r.release_reserved()628 if self.opt is not None:629 r.add_opt(self.opt, self.pad, opt_reserve, tsig_reserve)630 r.write_header()631 if self.tsig is not None:632 (new_tsig, ctx) = dns.tsig.sign(633 r.get_wire(),634 self.keyring,635 self.tsig[0],636 int(time.time()),637 self.request_mac,638 tsig_ctx,639 multi,640 )641 self.tsig.clear()642 self.tsig.add(new_tsig)643 r.add_rrset(dns.renderer.ADDITIONAL, self.tsig)644 r.write_header()645 if multi:646 self.tsig_ctx = ctx647 wire = r.get_wire()648 if prepend_length:649 wire = len(wire).to_bytes(2, "big") + wire650 return wire651 652 @staticmethod653 def _make_tsig(654 keyname, algorithm, time_signed, fudge, mac, original_id, error, other655 ):656 tsig = dns.rdtypes.ANY.TSIG.TSIG(657 dns.rdataclass.ANY,658 dns.rdatatype.TSIG,659 algorithm,660 time_signed,661 fudge,662 mac,663 original_id,664 error,665 other,666 )667 return dns.rrset.from_rdata(keyname, 0, tsig)668 669 def use_tsig(670 self,671 keyring: Any,672 keyname: Optional[Union[dns.name.Name, str]] = None,673 fudge: int = 300,674 original_id: Optional[int] = None,675 tsig_error: int = 0,676 other_data: bytes = b"",677 algorithm: Union[dns.name.Name, str] = dns.tsig.default_algorithm,678 ) -> None:679 """When sending, a TSIG signature using the specified key680 should be added.681 682 *key*, a ``dns.tsig.Key`` is the key to use. If a key is specified,683 the *keyring* and *algorithm* fields are not used.684 685 *keyring*, a ``dict``, ``callable`` or ``dns.tsig.Key``, is either686 the TSIG keyring or key to use.687 688 The format of a keyring dict is a mapping from TSIG key name, as689 ``dns.name.Name`` to ``dns.tsig.Key`` or a TSIG secret, a ``bytes``.690 If a ``dict`` *keyring* is specified but a *keyname* is not, the key691 used will be the first key in the *keyring*. Note that the order of692 keys in a dictionary is not defined, so applications should supply a693 keyname when a ``dict`` keyring is used, unless they know the keyring694 contains only one key. If a ``callable`` keyring is specified, the695 callable will be called with the message and the keyname, and is696 expected to return a key.697 698 *keyname*, a ``dns.name.Name``, ``str`` or ``None``, the name of699 this TSIG key to use; defaults to ``None``. If *keyring* is a700 ``dict``, the key must be defined in it. If *keyring* is a701 ``dns.tsig.Key``, this is ignored.702 703 *fudge*, an ``int``, the TSIG time fudge.704 705 *original_id*, an ``int``, the TSIG original id. If ``None``,706 the message's id is used.707 708 *tsig_error*, an ``int``, the TSIG error code.709 710 *other_data*, a ``bytes``, the TSIG other data.711 712 *algorithm*, a ``dns.name.Name`` or ``str``, the TSIG algorithm to use. This is713 only used if *keyring* is a ``dict``, and the key entry is a ``bytes``.714 """715 716 if isinstance(keyring, dns.tsig.Key):717 key = keyring718 keyname = key.name719 elif callable(keyring):720 key = keyring(self, keyname)721 else:722 if isinstance(keyname, str):723 keyname = dns.name.from_text(keyname)724 if keyname is None:725 keyname = next(iter(keyring))726 key = keyring[keyname]727 if isinstance(key, bytes):728 key = dns.tsig.Key(keyname, key, algorithm)729 self.keyring = key730 if original_id is None:731 original_id = self.id732 self.tsig = self._make_tsig(733 keyname,734 self.keyring.algorithm,735 0,736 fudge,737 b"\x00" * dns.tsig.mac_sizes[self.keyring.algorithm],738 original_id,739 tsig_error,740 other_data,741 )742 743 @property744 def keyname(self) -> Optional[dns.name.Name]:745 if self.tsig:746 return self.tsig.name747 else:748 return None749 750 @property751 def keyalgorithm(self) -> Optional[dns.name.Name]:752 if self.tsig:753 return self.tsig[0].algorithm754 else:755 return None756 757 @property758 def mac(self) -> Optional[bytes]:759 if self.tsig:760 return self.tsig[0].mac761 else:762 return None763 764 @property765 def tsig_error(self) -> Optional[int]:766 if self.tsig:767 return self.tsig[0].error768 else:769 return None770 771 @property772 def had_tsig(self) -> bool:773 return bool(self.tsig)774 775 @staticmethod776 def _make_opt(flags=0, payload=DEFAULT_EDNS_PAYLOAD, options=None):777 opt = dns.rdtypes.ANY.OPT.OPT(payload, dns.rdatatype.OPT, options or ())778 return dns.rrset.from_rdata(dns.name.root, int(flags), opt)779 780 def use_edns(781 self,782 edns: Optional[Union[int, bool]] = 0,783 ednsflags: int = 0,784 payload: int = DEFAULT_EDNS_PAYLOAD,785 request_payload: Optional[int] = None,786 options: Optional[List[dns.edns.Option]] = None,787 pad: int = 0,788 ) -> None:789 """Configure EDNS behavior.790 791 *edns*, an ``int``, is the EDNS level to use. Specifying ``None``, ``False``,792 or ``-1`` means "do not use EDNS", and in this case the other parameters are793 ignored. Specifying ``True`` is equivalent to specifying 0, i.e. "use EDNS0".794 795 *ednsflags*, an ``int``, the EDNS flag values.796 797 *payload*, an ``int``, is the EDNS sender's payload field, which is the maximum798 size of UDP datagram the sender can handle. I.e. how big a response to this799 message can be.800 801 *request_payload*, an ``int``, is the EDNS payload size to use when sending this802 message. If not specified, defaults to the value of *payload*.803 804 *options*, a list of ``dns.edns.Option`` objects or ``None``, the EDNS options.805 806 *pad*, a non-negative ``int``. If 0, the default, do not pad; otherwise add807 padding bytes to make the message size a multiple of *pad*. Note that if808 padding is non-zero, an EDNS PADDING option will always be added to the809 message.810 """811 812 if edns is None or edns is False:813 edns = -1814 elif edns is True:815 edns = 0816 if edns < 0:817 self.opt = None818 self.request_payload = 0819 else:820 # make sure the EDNS version in ednsflags agrees with edns821 ednsflags &= 0xFF00FFFF822 ednsflags |= edns << 16823 if options is None:824 options = []825 self.opt = self._make_opt(ednsflags, payload, options)826 if request_payload is None:827 request_payload = payload828 self.request_payload = request_payload829 if pad < 0:830 raise ValueError("pad must be non-negative")831 self.pad = pad832 833 @property834 def edns(self) -> int:835 if self.opt:836 return (self.ednsflags & 0xFF0000) >> 16837 else:838 return -1839 840 @property841 def ednsflags(self) -> int:842 if self.opt:843 return self.opt.ttl844 else:845 return 0846 847 @ednsflags.setter848 def ednsflags(self, v):849 if self.opt:850 self.opt.ttl = v851 elif v:852 self.opt = self._make_opt(v)853 854 @property855 def payload(self) -> int:856 if self.opt:857 return self.opt[0].payload858 else:859 return 0860 861 @property862 def options(self) -> Tuple:863 if self.opt:864 return self.opt[0].options865 else:866 return ()867 868 def want_dnssec(self, wanted: bool = True) -> None:869 """Enable or disable 'DNSSEC desired' flag in requests.870 871 *wanted*, a ``bool``. If ``True``, then DNSSEC data is872 desired in the response, EDNS is enabled if required, and then873 the DO bit is set. If ``False``, the DO bit is cleared if874 EDNS is enabled.875 """876 877 if wanted:878 self.ednsflags |= dns.flags.DO879 elif self.opt:880 self.ednsflags &= ~int(dns.flags.DO)881 882 def rcode(self) -> dns.rcode.Rcode:883 """Return the rcode.884 885 Returns a ``dns.rcode.Rcode``.886 """887 return dns.rcode.from_flags(int(self.flags), int(self.ednsflags))888 889 def set_rcode(self, rcode: dns.rcode.Rcode) -> None:890 """Set the rcode.891 892 *rcode*, a ``dns.rcode.Rcode``, is the rcode to set.893 """894 (value, evalue) = dns.rcode.to_flags(rcode)895 self.flags &= 0xFFF0896 self.flags |= value897 self.ednsflags &= 0x00FFFFFF898 self.ednsflags |= evalue899 900 def opcode(self) -> dns.opcode.Opcode:901 """Return the opcode.902 903 Returns a ``dns.opcode.Opcode``.904 """905 return dns.opcode.from_flags(int(self.flags))906 907 def set_opcode(self, opcode: dns.opcode.Opcode) -> None:908 """Set the opcode.909 910 *opcode*, a ``dns.opcode.Opcode``, is the opcode to set.911 """912 self.flags &= 0x87FF913 self.flags |= dns.opcode.to_flags(opcode)914 915 def _get_one_rr_per_rrset(self, value):916 # What the caller picked is fine.917 return value918 919 # pylint: disable=unused-argument920 921 def _parse_rr_header(self, section, name, rdclass, rdtype):922 return (rdclass, rdtype, None, False)923 924 # pylint: enable=unused-argument925 926 def _parse_special_rr_header(self, section, count, position, name, rdclass, rdtype):927 if rdtype == dns.rdatatype.OPT:928 if (929 section != MessageSection.ADDITIONAL930 or self.opt931 or name != dns.name.root932 ):933 raise BadEDNS934 elif rdtype == dns.rdatatype.TSIG:935 if (936 section != MessageSection.ADDITIONAL937 or rdclass != dns.rdatatype.ANY938 or position != count - 1939 ):940 raise BadTSIG941 return (rdclass, rdtype, None, False)942 943 944class ChainingResult:945 """The result of a call to dns.message.QueryMessage.resolve_chaining().946 947 The ``answer`` attribute is the answer RRSet, or ``None`` if it doesn't948 exist.949 950 The ``canonical_name`` attribute is the canonical name after all951 chaining has been applied (this is the same name as ``rrset.name`` in cases952 where rrset is not ``None``).953 954 The ``minimum_ttl`` attribute is the minimum TTL, i.e. the TTL to955 use if caching the data. It is the smallest of all the CNAME TTLs956 and either the answer TTL if it exists or the SOA TTL and SOA957 minimum values for negative answers.958 959 The ``cnames`` attribute is a list of all the CNAME RRSets followed to960 get to the canonical name.961 """962 963 def __init__(964 self,965 canonical_name: dns.name.Name,966 answer: Optional[dns.rrset.RRset],967 minimum_ttl: int,968 cnames: List[dns.rrset.RRset],969 ):970 self.canonical_name = canonical_name971 self.answer = answer972 self.minimum_ttl = minimum_ttl973 self.cnames = cnames974 975 976class QueryMessage(Message):977 def resolve_chaining(self) -> ChainingResult:978 """Follow the CNAME chain in the response to determine the answer979 RRset.980 981 Raises ``dns.message.NotQueryResponse`` if the message is not982 a response.983 984 Raises ``dns.message.ChainTooLong`` if the CNAME chain is too long.985 986 Raises ``dns.message.AnswerForNXDOMAIN`` if the rcode is NXDOMAIN987 but an answer was found.988 989 Raises ``dns.exception.FormError`` if the question count is not 1.990 991 Returns a ChainingResult object.992 """993 if self.flags & dns.flags.QR == 0:994 raise NotQueryResponse995 if len(self.question) != 1:996 raise dns.exception.FormError997 question = self.question[0]998 qname = question.name999 min_ttl = dns.ttl.MAX_TTL1000 answer = None1001 count = 01002 cnames = []1003 while count < MAX_CHAIN:1004 try:1005 answer = self.find_rrset(1006 self.answer, qname, question.rdclass, question.rdtype1007 )1008 min_ttl = min(min_ttl, answer.ttl)1009 break1010 except KeyError:1011 if question.rdtype != dns.rdatatype.CNAME:1012 try:1013 crrset = self.find_rrset(1014 self.answer, qname, question.rdclass, dns.rdatatype.CNAME1015 )1016 cnames.append(crrset)1017 min_ttl = min(min_ttl, crrset.ttl)1018 for rd in crrset:1019 qname = rd.target1020 break1021 count += 11022 continue1023 except KeyError:1024 # Exit the chaining loop1025 break1026 else:1027 # Exit the chaining loop1028 break1029 if count >= MAX_CHAIN:1030 raise ChainTooLong1031 if self.rcode() == dns.rcode.NXDOMAIN and answer is not None:1032 raise AnswerForNXDOMAIN1033 if answer is None:1034 # Further minimize the TTL with NCACHE.1035 auname = qname1036 while True:1037 # Look for an SOA RR whose owner name is a superdomain1038 # of qname.1039 try:1040 srrset = self.find_rrset(1041 self.authority, auname, question.rdclass, dns.rdatatype.SOA1042 )1043 min_ttl = min(min_ttl, srrset.ttl, srrset[0].minimum)1044 break1045 except KeyError:1046 try:1047 auname = auname.parent()1048 except dns.name.NoParent:1049 break1050 return ChainingResult(qname, answer, min_ttl, cnames)1051 1052 def canonical_name(self) -> dns.name.Name:1053 """Return the canonical name of the first name in the question1054 section.1055 1056 Raises ``dns.message.NotQueryResponse`` if the message is not1057 a response.1058 1059 Raises ``dns.message.ChainTooLong`` if the CNAME chain is too long.1060 1061 Raises ``dns.message.AnswerForNXDOMAIN`` if the rcode is NXDOMAIN1062 but an answer was found.1063 1064 Raises ``dns.exception.FormError`` if the question count is not 1.1065 """1066 return self.resolve_chaining().canonical_name1067 1068 1069def _maybe_import_update():1070 # We avoid circular imports by doing this here. We do it in another1071 # function as doing it in _message_factory_from_opcode() makes "dns"1072 # a local symbol, and the first line fails :)1073 1074 # pylint: disable=redefined-outer-name,import-outside-toplevel,unused-import1075 import dns.update # noqa: F4011076 1077 1078def _message_factory_from_opcode(opcode):1079 if opcode == dns.opcode.QUERY:1080 return QueryMessage1081 elif opcode == dns.opcode.UPDATE:1082 _maybe_import_update()1083 return dns.update.UpdateMessage1084 else:1085 return Message1086 1087 1088class _WireReader:1089 """Wire format reader.1090 1091 parser: the binary parser1092 message: The message object being built1093 initialize_message: Callback to set message parsing options1094 question_only: Are we only reading the question?1095 one_rr_per_rrset: Put each RR into its own RRset?1096 keyring: TSIG keyring1097 ignore_trailing: Ignore trailing junk at end of request?1098 multi: Is this message part of a multi-message sequence?1099 DNS dynamic updates.1100 continue_on_error: try to extract as much information as possible from1101 the message, accumulating MessageErrors in the *errors* attribute instead of1102 raising them.1103 """1104 1105 def __init__(1106 self,1107 wire,1108 initialize_message,1109 question_only=False,1110 one_rr_per_rrset=False,1111 ignore_trailing=False,1112 keyring=None,1113 multi=False,1114 continue_on_error=False,1115 ):1116 self.parser = dns.wire.Parser(wire)1117 self.message = None1118 self.initialize_message = initialize_message1119 self.question_only = question_only1120 self.one_rr_per_rrset = one_rr_per_rrset1121 self.ignore_trailing = ignore_trailing1122 self.keyring = keyring1123 self.multi = multi1124 self.continue_on_error = continue_on_error1125 self.errors = []1126 1127 def _get_question(self, section_number, qcount):1128 """Read the next *qcount* records from the wire data and add them to1129 the question section.1130 """1131 assert self.message is not None1132 section = self.message.sections[section_number]1133 for _ in range(qcount):1134 qname = self.parser.get_name(self.message.origin)1135 (rdtype, rdclass) = self.parser.get_struct("!HH")1136 (rdclass, rdtype, _, _) = self.message._parse_rr_header(1137 section_number, qname, rdclass, rdtype1138 )1139 self.message.find_rrset(1140 section, qname, rdclass, rdtype, create=True, force_unique=True1141 )1142 1143 def _add_error(self, e):1144 self.errors.append(MessageError(e, self.parser.current))1145 1146 def _get_section(self, section_number, count):1147 """Read the next I{count} records from the wire data and add them to1148 the specified section.1149 1150 section_number: the section of the message to which to add records1151 count: the number of records to read1152 """1153 assert self.message is not None1154 section = self.message.sections[section_number]1155 force_unique = self.one_rr_per_rrset1156 for i in range(count):1157 rr_start = self.parser.current1158 absolute_name = self.parser.get_name()1159 if self.message.origin is not None:1160 name = absolute_name.relativize(self.message.origin)1161 else:1162 name = absolute_name1163 (rdtype, rdclass, ttl, rdlen) = self.parser.get_struct("!HHIH")1164 if rdtype in (dns.rdatatype.OPT, dns.rdatatype.TSIG):1165 (1166 rdclass,1167 rdtype,1168 deleting,1169 empty,1170 ) = self.message._parse_special_rr_header(1171 section_number, count, i, name, rdclass, rdtype1172 )1173 else:1174 (rdclass, rdtype, deleting, empty) = self.message._parse_rr_header(1175 section_number, name, rdclass, rdtype1176 )1177 rdata_start = self.parser.current1178 try:1179 if empty:1180 if rdlen > 0:1181 raise dns.exception.FormError1182 rd = None1183 covers = dns.rdatatype.NONE1184 else:1185 with self.parser.restrict_to(rdlen):1186 rd = dns.rdata.from_wire_parser(1187 rdclass, rdtype, self.parser, self.message.origin1188 )1189 covers = rd.covers()1190 if self.message.xfr and rdtype == dns.rdatatype.SOA:1191 force_unique = True1192 if rdtype == dns.rdatatype.OPT:1193 self.message.opt = dns.rrset.from_rdata(name, ttl, rd)1194 elif rdtype == dns.rdatatype.TSIG:1195 if self.keyring is None:1196 raise UnknownTSIGKey("got signed message without keyring")1197 if isinstance(self.keyring, dict):1198 key = self.keyring.get(absolute_name)1199 if isinstance(key, bytes):1200 key = dns.tsig.Key(absolute_name, key, rd.algorithm)