Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/message.py: 19%
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
1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
3# Copyright (C) 2001-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.
18"""DNS Messages"""
20import dataclasses
21import enum
22import io
23import os
24import time
25from typing import Any, TextIO, cast
27import dns._file_util
28import dns.edns
29import dns.entropy
30import dns.enum
31import dns.exception
32import dns.flags
33import dns.name
34import dns.opcode
35import dns.rcode
36import dns.rdata
37import dns.rdataclass
38import dns.rdataset
39import dns.rdatatype
40import dns.rdtypes.ANY.OPT
41import dns.rdtypes.ANY.SOA
42import dns.rdtypes.ANY.TSIG
43import dns.renderer
44import dns.rrset
45import dns.tokenizer
46import dns.tsig
47import dns.ttl
48import dns.wire
51class ShortHeader(dns.exception.FormError):
52 """The DNS packet passed to from_wire() is too short."""
55class TrailingJunk(dns.exception.FormError):
56 """The DNS packet passed to from_wire() has extra junk at the end of it."""
59class UnknownHeaderField(dns.exception.DNSException):
60 """The header field name was not recognized when converting from text
61 into a message."""
64class BadEDNS(dns.exception.FormError):
65 """An OPT record occurred somewhere other than
66 the additional data section."""
69class BadTSIG(dns.exception.FormError):
70 """A TSIG record occurred somewhere other than the end of
71 the additional data section or the class is not ANY."""
74class UnknownTSIGKey(dns.exception.DNSException):
75 """A TSIG with an unknown key was received."""
78class Truncated(dns.exception.DNSException):
79 """The truncated flag is set."""
81 supp_kwargs = {"message"}
83 # We do this as otherwise mypy complains about unexpected keyword argument
84 # idna_exception
85 def __init__(self, *args, **kwargs):
86 super().__init__(*args, **kwargs)
88 def message(self):
89 """As much of the message as could be processed.
91 :rtype: :py:class:`dns.message.Message`
92 """
93 return self.kwargs["message"]
96class NotQueryResponse(dns.exception.DNSException):
97 """Message is not a response to a query."""
100class ChainTooLong(dns.exception.DNSException):
101 """The CNAME chain is too long."""
104class AnswerForNXDOMAIN(dns.exception.DNSException):
105 """The rcode is NXDOMAIN but an answer was found."""
108class NoPreviousName(dns.exception.SyntaxError):
109 """No previous name was known."""
112class MessageSection(dns.enum.IntEnum):
113 """Message sections"""
115 QUESTION = 0
116 ANSWER = 1
117 AUTHORITY = 2
118 ADDITIONAL = 3
120 @classmethod
121 def _maximum(cls):
122 return 3
125class MessageError:
126 def __init__(self, exception: Exception, offset: int):
127 self.exception = exception
128 self.offset = offset
131DEFAULT_EDNS_PAYLOAD = dns.renderer.DEFAULT_EDNS_PAYLOAD
132MAX_CHAIN = 16
134IndexKeyType = tuple[
135 int,
136 dns.name.Name,
137 dns.rdataclass.RdataClass,
138 dns.rdatatype.RdataType,
139 dns.rdatatype.RdataType | None,
140 dns.rdataclass.RdataClass | None,
141]
142IndexType = dict[IndexKeyType, dns.rrset.RRset]
143SectionType = int | str | list[dns.rrset.RRset]
146@dataclasses.dataclass(frozen=True)
147class MessageStyle(dns.rdataset.RdatasetStyle):
148 """Message text styles.
150 A ``MessageStyle`` is also a :py:class:`dns.name.NameStyle` and a
151 :py:class:`dns.rdata.RdataStyle`, and a :py:class:`dns.rdataset.RdatasetStyle`.
152 See those classes for a description of their options.
154 There are currently no message-specific style options, but if that changes they
155 will be documented here.
156 """
159class Message:
160 """A DNS message."""
162 _section_enum = MessageSection
164 def __init__(self, id: int | None = None):
165 if id is None:
166 self.id = dns.entropy.random_16()
167 else:
168 self.id = id
169 self.flags = 0
170 self.sections: list[list[dns.rrset.RRset]] = [[], [], [], []]
171 self.opt: dns.rrset.RRset | None = None
172 self.request_payload = 0
173 self.pad = 0
174 self.keyring: Any = None
175 self.tsig: dns.rrset.RRset | None = None
176 self.want_tsig_sign = False
177 self.request_mac = b""
178 self.xfr = False
179 self.origin: dns.name.Name | None = None
180 self.tsig_ctx: Any | None = None
181 self.index: IndexType = {}
182 self.errors: list[MessageError] = []
183 self.time = 0.0
184 self.wire: bytes | None = None
186 @property
187 def question(self) -> list[dns.rrset.RRset]:
188 """The question section."""
189 return self.sections[0]
191 @question.setter
192 def question(self, v):
193 self.sections[0] = v
195 @property
196 def answer(self) -> list[dns.rrset.RRset]:
197 """The answer section."""
198 return self.sections[1]
200 @answer.setter
201 def answer(self, v):
202 self.sections[1] = v
204 @property
205 def authority(self) -> list[dns.rrset.RRset]:
206 """The authority section."""
207 return self.sections[2]
209 @authority.setter
210 def authority(self, v):
211 self.sections[2] = v
213 @property
214 def additional(self) -> list[dns.rrset.RRset]:
215 """The additional data section."""
216 return self.sections[3]
218 @additional.setter
219 def additional(self, v):
220 self.sections[3] = v
222 def __repr__(self):
223 return "<DNS message, ID " + repr(self.id) + ">"
225 def __str__(self):
226 return self.to_text()
228 def to_text(
229 self,
230 origin: dns.name.Name | None = None,
231 relativize: bool = True,
232 style: MessageStyle | None = None,
233 **kw: Any,
234 ) -> str:
235 """Convert the message to text.
237 The *origin*, *relativize*, and any other keyword
238 arguments are passed to the RRset ``to_text()`` method.
240 :param style: If specified, overrides *origin* and *relativize*.
241 :type style: :py:class:`dns.rdataset.RdatasetStyle` or ``None``
242 :rtype: str
243 """
244 if style is None:
245 kw = kw.copy()
246 kw["origin"] = origin
247 kw["relativize"] = relativize
248 style = MessageStyle.from_keywords(kw)
249 return self.to_styled_text(style)
251 def to_styled_text(self, style: MessageStyle) -> str:
252 """Convert the message to styled text.
254 :param style: The style to use for formatting.
255 :type style: :py:class:`dns.message.MessageStyle`
256 :rtype: str
257 """
259 s = io.StringIO()
260 s.write(f"id {self.id}\n")
261 s.write(f"opcode {dns.opcode.to_text(self.opcode())}\n")
262 s.write(f"rcode {dns.rcode.to_text(self.rcode())}\n")
263 s.write(f"flags {dns.flags.to_text(self.flags)}\n")
264 if self.edns >= 0:
265 s.write(f"edns {self.edns}\n")
266 if self.ednsflags != 0:
267 s.write(f"eflags {dns.flags.edns_to_text(self.ednsflags)}\n")
268 s.write(f"payload {self.payload}\n")
269 for opt in self.options:
270 s.write(f"option {opt.to_text()}\n")
271 for name, which in self._section_enum.__members__.items():
272 s.write(f";{name}\n")
273 for rrset in self.section_from_number(which):
274 s.write(rrset.to_styled_text(style))
275 s.write("\n")
276 if self.tsig is not None:
277 s.write(self.tsig.to_styled_text(style))
278 s.write("\n")
279 #
280 # We strip off the final \n so the caller can print the result without
281 # doing weird things to get around eccentricities in Python print
282 # formatting
283 #
284 return s.getvalue()[:-1]
286 def __eq__(self, other):
287 """Two messages are equal if they have the same content in the
288 header, question, answer, and authority sections.
290 :rtype: bool
291 """
293 if not isinstance(other, Message):
294 return False
295 if self.id != other.id:
296 return False
297 if self.flags != other.flags:
298 return False
299 for i, section in enumerate(self.sections):
300 other_section = other.sections[i]
301 for n in section:
302 if n not in other_section:
303 return False
304 for n in other_section:
305 if n not in section:
306 return False
307 return True
309 def __ne__(self, other):
310 return not self.__eq__(other)
312 def is_response(self, other: "Message") -> bool:
313 """Is *other* a response to this message?
315 :param other: The message to check.
316 :type other: :py:class:`dns.message.Message`
317 :rtype: bool
318 """
320 if (
321 other.flags & dns.flags.QR == 0
322 or self.id != other.id
323 or dns.opcode.from_flags(self.flags) != dns.opcode.from_flags(other.flags)
324 ):
325 return False
326 if other.rcode() in {
327 dns.rcode.FORMERR,
328 dns.rcode.SERVFAIL,
329 dns.rcode.NOTIMP,
330 dns.rcode.REFUSED,
331 }:
332 # We don't check the question section in these cases if
333 # the other question section is empty, even though they
334 # still really ought to have a question section.
335 if len(other.question) == 0:
336 return True
337 if dns.opcode.is_update(self.flags):
338 # This is assuming the "sender doesn't include anything
339 # from the update", but we don't care to check the other
340 # case, which is that all the sections are returned and
341 # identical.
342 return True
343 for n in self.question:
344 if n not in other.question:
345 return False
346 for n in other.question:
347 if n not in self.question:
348 return False
349 return True
351 def section_number(self, section: list[dns.rrset.RRset]) -> int:
352 """Return the "section number" of the specified section for use
353 in indexing.
355 :param section: One of the section attributes of this message.
356 :raises ValueError: If the section is not known.
357 :rtype: int
358 """
360 for i, our_section in enumerate(self.sections):
361 if section is our_section:
362 return self._section_enum(i)
363 raise ValueError("unknown section")
365 def section_from_number(self, number: int) -> list[dns.rrset.RRset]:
366 """Return the section list associated with the specified section
367 number.
369 :param number: A section number (``int``) or text name of a section.
370 :raises ValueError: If the section is not known.
371 :rtype: list
372 """
374 section = self._section_enum.make(number)
375 return self.sections[section]
377 def find_rrset(
378 self,
379 section: SectionType,
380 name: dns.name.Name,
381 rdclass: dns.rdataclass.RdataClass,
382 rdtype: dns.rdatatype.RdataType,
383 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
384 deleting: dns.rdataclass.RdataClass | None = None,
385 create: bool = False,
386 force_unique: bool = False,
387 idna_codec: dns.name.IDNACodec | None = None,
388 ) -> dns.rrset.RRset:
389 """Find the RRset with the given attributes in the specified section.
391 *section* may be an ``int`` section number, a ``str`` section name, or
392 one of the section list attributes of this message. For example::
394 my_message.find_rrset(my_message.answer, name, rdclass, rdtype)
395 my_message.find_rrset(dns.message.ANSWER, name, rdclass, rdtype)
396 my_message.find_rrset("ANSWER", name, rdclass, rdtype)
398 :param name: The owner name.
399 :type name: :py:class:`dns.name.Name` or ``str``
400 :param rdclass: The rdata class.
401 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
402 :param rdtype: The rdata type.
403 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
404 :param covers: The covered type; default is ``dns.rdatatype.NONE``.
405 :type covers: :py:class:`dns.rdatatype.RdataType` or ``str``
406 :param deleting: The deleting value; default is ``None``.
407 :type deleting: :py:class:`dns.rdataclass.RdataClass`, ``str``, or ``None``
408 :param create: If ``True``, create and append the RRset if not found.
409 :type create: bool
410 :param force_unique: If ``True`` and *create* is ``True``, always create
411 a new RRset even if a matching one exists. Useful for DDNS updates.
412 :type force_unique: bool
413 :param idna_codec: The IDNA encoder/decoder. Defaults to IDNA 2003.
414 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
415 :raises KeyError: If the RRset was not found and *create* is ``False``.
416 :rtype: :py:class:`dns.rrset.RRset`
417 """
419 if isinstance(section, int):
420 section_number = section
421 section = self.section_from_number(section_number)
422 elif isinstance(section, str):
423 section_number = self._section_enum.from_text(section)
424 section = self.section_from_number(section_number)
425 else:
426 section_number = self.section_number(section)
427 if isinstance(name, str):
428 name = dns.name.from_text(name, idna_codec=idna_codec)
429 rdtype = dns.rdatatype.RdataType.make(rdtype)
430 rdclass = dns.rdataclass.RdataClass.make(rdclass)
431 covers = dns.rdatatype.RdataType.make(covers)
432 if deleting is not None:
433 deleting = dns.rdataclass.RdataClass.make(deleting)
434 key = (section_number, name, rdclass, rdtype, covers, deleting)
435 if not force_unique:
436 if self.index is not None:
437 rrset = self.index.get(key)
438 if rrset is not None:
439 return rrset
440 else:
441 for rrset in section:
442 if rrset.full_match(name, rdclass, rdtype, covers, deleting):
443 return rrset
444 if not create:
445 raise KeyError
446 rrset = dns.rrset.RRset(name, rdclass, rdtype, covers, deleting)
447 section.append(rrset)
448 if self.index is not None:
449 self.index[key] = rrset
450 return rrset
452 def get_rrset(
453 self,
454 section: SectionType,
455 name: dns.name.Name,
456 rdclass: dns.rdataclass.RdataClass,
457 rdtype: dns.rdatatype.RdataType,
458 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
459 deleting: dns.rdataclass.RdataClass | None = None,
460 create: bool = False,
461 force_unique: bool = False,
462 idna_codec: dns.name.IDNACodec | None = None,
463 ) -> dns.rrset.RRset | None:
464 """Get the RRset with the given attributes in the specified section.
466 Like :py:meth:`find_rrset` but returns ``None`` instead of raising
467 :py:exc:`KeyError` when the RRset is not found.
469 *section* may be an ``int`` section number, a ``str`` section name, or
470 one of the section list attributes of this message.
472 :param name: The owner name.
473 :type name: :py:class:`dns.name.Name` or ``str``
474 :param rdclass: The rdata class.
475 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
476 :param rdtype: The rdata type.
477 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
478 :param covers: The covered type; default is ``dns.rdatatype.NONE``.
479 :type covers: :py:class:`dns.rdatatype.RdataType` or ``str``
480 :param deleting: The deleting value; default is ``None``.
481 :type deleting: :py:class:`dns.rdataclass.RdataClass`, ``str``, or ``None``
482 :param create: If ``True``, create and append the RRset if not found.
483 :type create: bool
484 :param force_unique: If ``True`` and *create* is ``True``, always create
485 a new RRset even if a matching one exists.
486 :type force_unique: bool
487 :param idna_codec: The IDNA encoder/decoder. Defaults to IDNA 2003.
488 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
489 :returns: The matching RRset, or ``None`` if not found.
490 :rtype: :py:class:`dns.rrset.RRset` or ``None``
491 """
493 try:
494 rrset = self.find_rrset(
495 section,
496 name,
497 rdclass,
498 rdtype,
499 covers,
500 deleting,
501 create,
502 force_unique,
503 idna_codec,
504 )
505 except KeyError:
506 rrset = None
507 return rrset
509 def section_count(self, section: SectionType) -> int:
510 """Returns the number of records in the specified section.
512 :param section: An ``int`` section number, a ``str`` section name, or
513 one of the section attributes of this message. For example::
515 my_message.section_count(my_message.answer)
516 my_message.section_count(dns.message.ANSWER)
517 my_message.section_count("ANSWER")
518 """
520 if isinstance(section, int):
521 section_number = section
522 section = self.section_from_number(section_number)
523 elif isinstance(section, str):
524 section_number = self._section_enum.from_text(section)
525 section = self.section_from_number(section_number)
526 else:
527 section_number = self.section_number(section)
528 count = sum(max(1, len(rrs)) for rrs in section)
529 if section_number == MessageSection.ADDITIONAL:
530 if self.opt is not None:
531 count += 1
532 if self.tsig is not None:
533 count += 1
534 return count
536 def _compute_opt_reserve(self) -> int:
537 """Compute the size required for the OPT RR, padding excluded"""
538 if not self.opt:
539 return 0
540 # 1 byte for the root name, 10 for the standard RR fields
541 size = 11
542 # This would be more efficient if options had a size() method, but we won't
543 # worry about that for now. We also don't worry if there is an existing padding
544 # option, as it is unlikely and probably harmless, as the worst case is that we
545 # may add another, and this seems to be legal.
546 opt_rdata = cast(dns.rdtypes.ANY.OPT.OPT, self.opt[0])
547 for option in opt_rdata.options:
548 wire = option.to_wire()
549 # We add 4 here to account for the option type and length
550 size += len(wire) + 4
551 if self.pad:
552 # Padding will be added, so again add the option type and length.
553 size += 4
554 return size
556 def _compute_tsig_reserve(self) -> int:
557 """Compute the size required for the TSIG RR"""
558 # This would be more efficient if TSIGs had a size method, but we won't
559 # worry about for now. Also, we can't really cope with the potential
560 # compressibility of the TSIG owner name, so we estimate with the uncompressed
561 # size. We will disable compression when TSIG and padding are both is active
562 # so that the padding comes out right.
563 if not self.tsig:
564 return 0
565 f = io.BytesIO()
566 self.tsig.to_wire(f)
567 return len(f.getvalue())
569 def to_wire(
570 self,
571 origin: dns.name.Name | None = None,
572 max_size: int = 0,
573 multi: bool = False,
574 tsig_ctx: Any | None = None,
575 prepend_length: bool = False,
576 prefer_truncation: bool = False,
577 **kw: Any,
578 ) -> bytes:
579 """Return the message in DNS compressed wire format.
581 Additional keyword arguments are passed to the RRset ``to_wire()``
582 method.
584 :param origin: Origin to append to relative names. If ``None``, the
585 message's own origin (if any) is used.
586 :type origin: :py:class:`dns.name.Name` or ``None``
587 :param max_size: Maximum wire format size; 0 means use the request
588 payload or 65535.
589 :type max_size: int
590 :param multi: ``True`` if this message is part of a multi-message sequence.
591 :type multi: bool
592 :param tsig_ctx: Ongoing TSIG context for zone transfer signing.
593 :param prepend_length: If ``True``, prepend the 2-byte message length
594 (useful for TCP/TLS/QUIC).
595 :type prepend_length: bool
596 :param prefer_truncation: If ``True``, truncate instead of raising when
597 the message exceeds *max_size*. Sets TC if truncation is before the
598 additional section.
599 :type prefer_truncation: bool
600 :raises dns.exception.TooBig: If *max_size* is exceeded and
601 *prefer_truncation* is ``False``.
602 :rtype: bytes
603 """
605 if origin is None and self.origin is not None:
606 origin = self.origin
607 if max_size == 0:
608 if self.request_payload != 0:
609 max_size = self.request_payload
610 else:
611 max_size = 65535
612 if max_size < 512:
613 max_size = 512
614 elif max_size > 65535:
615 max_size = 65535
616 r = dns.renderer.Renderer(self.id, self.flags, max_size, origin)
617 opt_reserve = self._compute_opt_reserve()
618 r.reserve(opt_reserve)
619 tsig_reserve = self._compute_tsig_reserve()
620 r.reserve(tsig_reserve)
621 try:
622 for rrset in self.question:
623 r.add_question(rrset.name, rrset.rdtype, rrset.rdclass)
624 for rrset in self.answer:
625 r.add_rrset(dns.renderer.ANSWER, rrset, **kw)
626 for rrset in self.authority:
627 r.add_rrset(dns.renderer.AUTHORITY, rrset, **kw)
628 for rrset in self.additional:
629 r.add_rrset(dns.renderer.ADDITIONAL, rrset, **kw)
630 except dns.exception.TooBig:
631 if prefer_truncation:
632 if r.section < dns.renderer.ADDITIONAL:
633 r.flags |= dns.flags.TC
634 else:
635 raise
636 r.release_reserved()
637 if self.opt is not None:
638 r.add_opt(self.opt, self.pad, opt_reserve, tsig_reserve)
639 r.write_header()
640 if self.tsig is not None:
641 if self.want_tsig_sign:
642 new_tsig, ctx = dns.tsig.sign(
643 r.get_wire(),
644 self.keyring,
645 self.tsig[0],
646 int(time.time()),
647 self.request_mac,
648 tsig_ctx,
649 multi,
650 )
651 self.tsig.clear()
652 self.tsig.add(new_tsig)
653 if multi:
654 self.tsig_ctx = ctx
655 r.add_rrset(dns.renderer.ADDITIONAL, self.tsig)
656 r.write_header()
657 wire = r.get_wire()
658 self.wire = wire
659 if prepend_length:
660 wire = len(wire).to_bytes(2, "big") + wire
661 return wire
663 @staticmethod
664 def _make_tsig(
665 keyname, algorithm, time_signed, fudge, mac, original_id, error, other
666 ):
667 return dns.renderer._make_tsig(
668 keyname, algorithm, time_signed, fudge, mac, original_id, error, other
669 )
671 def use_tsig(
672 self,
673 keyring: Any,
674 keyname: dns.name.Name | str | None = None,
675 fudge: int = 300,
676 original_id: int | None = None,
677 tsig_error: int = 0,
678 other_data: bytes = b"",
679 algorithm: dns.name.Name | str = dns.tsig.default_algorithm,
680 ) -> None:
681 """Arrange for a TSIG signature to be added when sending.
683 :param keyring: The TSIG keyring or key. A ``dict`` maps
684 :py:class:`dns.name.Name` keys to :py:class:`dns.tsig.Key` objects
685 or ``bytes`` secrets. If a ``dict`` is given without *keyname*, the
686 first key in the dict is used. A callable is invoked with the message
687 and *keyname* and must return a key.
688 :type keyring: dict, callable, or :py:class:`dns.tsig.Key`
689 :param keyname: The TSIG key name. Ignored if *keyring* is a
690 :py:class:`dns.tsig.Key`. Defaults to ``None``.
691 :type keyname: :py:class:`dns.name.Name`, ``str``, or ``None``
692 :param fudge: The TSIG time fudge.
693 :type fudge: int
694 :param original_id: The TSIG original id. Defaults to the message id.
695 :type original_id: int or ``None``
696 :param tsig_error: The TSIG error code.
697 :type tsig_error: int
698 :param other_data: The TSIG other data.
699 :type other_data: bytes
700 :param algorithm: The TSIG algorithm. Only used when *keyring* is a
701 ``dict`` and the key entry is ``bytes``.
702 :type algorithm: :py:class:`dns.name.Name` or ``str``
703 """
705 if isinstance(keyring, dns.tsig.Key):
706 key = keyring
707 keyname = key.name
708 elif callable(keyring):
709 key = keyring(self, keyname)
710 else:
711 if isinstance(keyname, str):
712 keyname = dns.name.from_text(keyname)
713 if keyname is None:
714 keyname = next(iter(keyring))
715 key = keyring[keyname]
716 if isinstance(key, bytes):
717 key = dns.tsig.Key(keyname, key, algorithm)
718 self.keyring = key
719 if original_id is None:
720 original_id = self.id
721 self.tsig = self._make_tsig(
722 keyname,
723 self.keyring.algorithm,
724 0,
725 fudge,
726 b"\x00" * dns.tsig.mac_sizes[self.keyring.algorithm],
727 original_id,
728 tsig_error,
729 other_data,
730 )
731 self.want_tsig_sign = True
733 @property
734 def keyname(self) -> dns.name.Name | None:
735 if self.tsig:
736 return self.tsig.name
737 else:
738 return None
740 @property
741 def keyalgorithm(self) -> dns.name.Name | None:
742 if self.tsig:
743 rdata = cast(dns.rdtypes.ANY.TSIG.TSIG, self.tsig[0])
744 return rdata.algorithm
745 else:
746 return None
748 @property
749 def mac(self) -> bytes | None:
750 if self.tsig:
751 rdata = cast(dns.rdtypes.ANY.TSIG.TSIG, self.tsig[0])
752 return rdata.mac
753 else:
754 return None
756 @property
757 def tsig_error(self) -> int | None:
758 if self.tsig:
759 rdata = cast(dns.rdtypes.ANY.TSIG.TSIG, self.tsig[0])
760 return rdata.error
761 else:
762 return None
764 @property
765 def had_tsig(self) -> bool:
766 return bool(self.tsig)
768 @staticmethod
769 def _make_opt(flags=0, payload=DEFAULT_EDNS_PAYLOAD, options=None):
770 return dns.renderer._make_opt(flags, payload, options)
772 def use_edns(
773 self,
774 edns: int | bool | None = 0,
775 ednsflags: int = 0,
776 payload: int = DEFAULT_EDNS_PAYLOAD,
777 request_payload: int | None = None,
778 options: list[dns.edns.Option] | None = None,
779 pad: int = 0,
780 ) -> None:
781 """Configure EDNS behavior.
783 :param edns: The EDNS level to use. Specifying ``None``, ``False``, or
784 ``-1`` means "do not use EDNS" (other parameters are ignored).
785 Specifying ``True`` is equivalent to specifying 0 (use EDNS0).
786 :type edns: int or ``None``
787 :param ednsflags: The EDNS flag values.
788 :type ednsflags: int
789 :param payload: The EDNS sender's payload field — the maximum UDP
790 datagram size the sender can handle (i.e. how big a response can be).
791 :type payload: int
792 :param request_payload: The EDNS payload size to use when sending.
793 Defaults to the value of *payload*.
794 :type request_payload: int or ``None``
795 :param options: The EDNS options.
796 :type options: list of :py:class:`dns.edns.Option` or ``None``
797 :param pad: If 0 (the default), do not pad; otherwise add padding bytes
798 to make the message size a multiple of *pad*. When nonzero, an EDNS
799 PADDING option is always added.
800 :type pad: int
801 """
803 if edns is None or edns is False:
804 edns = -1
805 elif edns is True:
806 edns = 0
807 if edns < 0:
808 self.opt = None
809 self.request_payload = 0
810 else:
811 # make sure the EDNS version in ednsflags agrees with edns
812 ednsflags &= 0xFF00FFFF
813 ednsflags |= edns << 16
814 if options is None:
815 options = []
816 self.opt = self._make_opt(ednsflags, payload, options)
817 if request_payload is None:
818 request_payload = payload
819 self.request_payload = request_payload
820 if pad < 0:
821 raise ValueError("pad must be non-negative")
822 self.pad = pad
824 @property
825 def edns(self) -> int:
826 if self.opt:
827 return (self.ednsflags & 0xFF0000) >> 16
828 else:
829 return -1
831 @property
832 def ednsflags(self) -> int:
833 if self.opt:
834 return self.opt.ttl
835 else:
836 return 0
838 @ednsflags.setter
839 def ednsflags(self, v):
840 if self.opt:
841 self.opt.ttl = v
842 elif v:
843 self.opt = self._make_opt(v)
845 @property
846 def payload(self) -> int:
847 if self.opt:
848 rdata = cast(dns.rdtypes.ANY.OPT.OPT, self.opt[0])
849 return rdata.payload
850 else:
851 return 0
853 @property
854 def options(self) -> tuple:
855 if self.opt:
856 rdata = cast(dns.rdtypes.ANY.OPT.OPT, self.opt[0])
857 return rdata.options
858 else:
859 return ()
861 def want_dnssec(self, wanted: bool = True) -> None:
862 """Enable or disable 'DNSSEC desired' flag in requests.
864 :param wanted: If ``True``, DNSSEC data is desired in the response,
865 EDNS is enabled if required, and the DO bit is set. If ``False``,
866 the DO bit is cleared if EDNS is enabled.
867 :type wanted: bool
868 """
870 if wanted:
871 self.ednsflags |= dns.flags.DO
872 elif self.opt:
873 self.ednsflags &= ~int(dns.flags.DO)
875 def rcode(self) -> dns.rcode.Rcode:
876 """Return the rcode.
878 :rtype: :py:class:`dns.rcode.Rcode`
879 """
880 return dns.rcode.from_flags(int(self.flags), int(self.ednsflags))
882 def set_rcode(self, rcode: dns.rcode.Rcode) -> None:
883 """Set the rcode.
885 *rcode*, a ``dns.rcode.Rcode``, is the rcode to set.
886 """
887 value, evalue = dns.rcode.to_flags(rcode)
888 self.flags &= 0xFFF0
889 self.flags |= value
890 self.ednsflags &= 0x00FFFFFF
891 self.ednsflags |= evalue
893 def opcode(self) -> dns.opcode.Opcode:
894 """Return the opcode.
896 :rtype: :py:class:`dns.opcode.Opcode`
897 """
898 return dns.opcode.from_flags(int(self.flags))
900 def set_opcode(self, opcode: dns.opcode.Opcode) -> None:
901 """Set the opcode.
903 :param opcode: The opcode to set.
904 :type opcode: :py:class:`dns.opcode.Opcode`
905 """
906 self.flags &= 0x87FF
907 self.flags |= dns.opcode.to_flags(opcode)
909 def get_options(self, otype: dns.edns.OptionType) -> list[dns.edns.Option]:
910 """Return the list of options of the specified type."""
911 return [option for option in self.options if option.otype == otype]
913 def extended_errors(self) -> list[dns.edns.EDEOption]:
914 """Return the list of Extended DNS Error (EDE) options in the message"""
915 return cast(list[dns.edns.EDEOption], self.get_options(dns.edns.OptionType.EDE))
917 def _get_one_rr_per_rrset(self, value):
918 # What the caller picked is fine.
919 return value
921 # pylint: disable=unused-argument
923 def _parse_rr_header(self, section, name, rdclass, rdtype):
924 return (rdclass, rdtype, None, False)
926 # pylint: enable=unused-argument
928 def _parse_special_rr_header(self, section, count, position, name, rdclass, rdtype):
929 if rdtype == dns.rdatatype.OPT:
930 if (
931 section != MessageSection.ADDITIONAL
932 or self.opt
933 or name != dns.name.root
934 ):
935 raise BadEDNS
936 elif rdtype == dns.rdatatype.TSIG:
937 if (
938 section != MessageSection.ADDITIONAL
939 or rdclass != dns.rdatatype.ANY
940 or position != count - 1
941 ):
942 raise BadTSIG
943 return (rdclass, rdtype, None, False)
946class ChainingResult:
947 """The result of a call to dns.message.QueryMessage.resolve_chaining().
949 The ``answer`` attribute is the answer RRSet, or ``None`` if it doesn't
950 exist.
952 The ``canonical_name`` attribute is the canonical name after all
953 chaining has been applied (this is the same name as ``rrset.name`` in cases
954 where rrset is not ``None``).
956 The ``minimum_ttl`` attribute is the minimum TTL, i.e. the TTL to
957 use if caching the data. It is the smallest of all the CNAME TTLs
958 and either the answer TTL if it exists or the SOA TTL and SOA
959 minimum values for negative answers.
961 The ``cnames`` attribute is a list of all the CNAME RRSets followed to
962 get to the canonical name.
963 """
965 def __init__(
966 self,
967 canonical_name: dns.name.Name,
968 answer: dns.rrset.RRset | None,
969 minimum_ttl: int,
970 cnames: list[dns.rrset.RRset],
971 ):
972 self.canonical_name = canonical_name
973 self.answer = answer
974 self.minimum_ttl = minimum_ttl
975 self.cnames = cnames
978class QueryMessage(Message):
979 def resolve_chaining(self) -> ChainingResult:
980 """Follow the CNAME chain in the response to determine the answer
981 RRset.
983 :raises dns.message.NotQueryResponse: If the message is not a response.
984 :raises dns.message.ChainTooLong: If the CNAME chain is too long.
985 :raises dns.message.AnswerForNXDOMAIN: If the rcode is NXDOMAIN but
986 an answer was found.
987 :raises dns.exception.FormError: If the question count is not 1.
988 :rtype: :py:class:`dns.message.ChainingResult`
989 """
990 if self.flags & dns.flags.QR == 0:
991 raise NotQueryResponse
992 if len(self.question) != 1:
993 raise dns.exception.FormError
994 question = self.question[0]
995 qname = question.name
996 min_ttl = dns.ttl.MAX_TTL
997 answer = None
998 count = 0
999 cnames = []
1000 while count < MAX_CHAIN:
1001 try:
1002 answer = self.find_rrset(
1003 self.answer, qname, question.rdclass, question.rdtype
1004 )
1005 min_ttl = min(min_ttl, answer.ttl)
1006 break
1007 except KeyError:
1008 if question.rdtype != dns.rdatatype.CNAME:
1009 try:
1010 crrset = self.find_rrset(
1011 self.answer, qname, question.rdclass, dns.rdatatype.CNAME
1012 )
1013 cnames.append(crrset)
1014 min_ttl = min(min_ttl, crrset.ttl)
1015 for rd in crrset:
1016 qname = rd.target
1017 break
1018 count += 1
1019 continue
1020 except KeyError:
1021 # Exit the chaining loop
1022 break
1023 else:
1024 # Exit the chaining loop
1025 break
1026 if count >= MAX_CHAIN:
1027 raise ChainTooLong
1028 if self.rcode() == dns.rcode.NXDOMAIN and answer is not None:
1029 raise AnswerForNXDOMAIN
1030 if answer is None:
1031 # Further minimize the TTL with NCACHE.
1032 auname = qname
1033 while True:
1034 # Look for an SOA RR whose owner name is a superdomain
1035 # of qname.
1036 try:
1037 srrset = self.find_rrset(
1038 self.authority, auname, question.rdclass, dns.rdatatype.SOA
1039 )
1040 srdata = cast(dns.rdtypes.ANY.SOA.SOA, srrset[0])
1041 min_ttl = min(min_ttl, srrset.ttl, srdata.minimum)
1042 break
1043 except KeyError:
1044 try:
1045 auname = auname.parent()
1046 except dns.name.NoParent:
1047 break
1048 return ChainingResult(qname, answer, min_ttl, cnames)
1050 def canonical_name(self) -> dns.name.Name:
1051 """Return the canonical name of the first name in the question
1052 section.
1054 :raises dns.message.NotQueryResponse: If the message is not a response.
1055 :raises dns.message.ChainTooLong: If the CNAME chain is too long.
1056 :raises dns.message.AnswerForNXDOMAIN: If the rcode is NXDOMAIN but
1057 an answer was found.
1058 :raises dns.exception.FormError: If the question count is not 1.
1059 """
1060 return self.resolve_chaining().canonical_name
1063def _maybe_import_update():
1064 # We avoid circular imports by doing this here. We do it in another
1065 # function as doing it in _message_factory_from_opcode() makes "dns"
1066 # a local symbol, and the first line fails :)
1068 # pylint: disable=redefined-outer-name,import-outside-toplevel,unused-import
1069 import dns.update # noqa: F401
1072def _message_factory_from_opcode(opcode):
1073 if opcode == dns.opcode.QUERY:
1074 return QueryMessage
1075 elif opcode == dns.opcode.UPDATE:
1076 _maybe_import_update()
1077 return dns.update.UpdateMessage # pyright: ignore
1078 else:
1079 return Message
1082class _WireReader:
1083 """Wire format reader.
1085 parser: the binary parser
1086 message: The message object being built
1087 initialize_message: Callback to set message parsing options
1088 question_only: Are we only reading the question?
1089 one_rr_per_rrset: Put each RR into its own RRset?
1090 keyring: TSIG keyring
1091 ignore_trailing: Ignore trailing junk at end of request?
1092 multi: Is this message part of a multi-message sequence?
1093 DNS dynamic updates.
1094 continue_on_error: try to extract as much information as possible from
1095 the message, accumulating MessageErrors in the *errors* attribute instead of
1096 raising them.
1097 """
1099 def __init__(
1100 self,
1101 wire,
1102 initialize_message,
1103 question_only=False,
1104 one_rr_per_rrset=False,
1105 ignore_trailing=False,
1106 keyring=None,
1107 multi=False,
1108 continue_on_error=False,
1109 ):
1110 self.parser = dns.wire.Parser(wire)
1111 self.message = None
1112 self.initialize_message = initialize_message
1113 self.question_only = question_only
1114 self.one_rr_per_rrset = one_rr_per_rrset
1115 self.ignore_trailing = ignore_trailing
1116 self.keyring = keyring
1117 self.multi = multi
1118 self.continue_on_error = continue_on_error
1119 self.errors = []
1121 def _get_question(self, section_number, qcount):
1122 """Read the next *qcount* records from the wire data and add them to
1123 the question section.
1124 """
1125 assert self.message is not None
1126 section = self.message.sections[section_number]
1127 for _ in range(qcount):
1128 qname = self.parser.get_name(self.message.origin)
1129 rdtype, rdclass = self.parser.get_struct("!HH")
1130 rdclass, rdtype, _, _ = self.message._parse_rr_header(
1131 section_number, qname, rdclass, rdtype
1132 )
1133 self.message.find_rrset(
1134 section, qname, rdclass, rdtype, create=True, force_unique=True
1135 )
1137 def _add_error(self, e):
1138 self.errors.append(MessageError(e, self.parser.current))
1140 def _get_section(self, section_number, count):
1141 """Read the next I{count} records from the wire data and add them to
1142 the specified section.
1144 section_number: the section of the message to which to add records
1145 count: the number of records to read
1146 """
1147 assert self.message is not None
1148 section = self.message.sections[section_number]
1149 force_unique = self.one_rr_per_rrset
1150 for i in range(count):
1151 rr_start = self.parser.current
1152 absolute_name = self.parser.get_name()
1153 if self.message.origin is not None:
1154 name = absolute_name.relativize(self.message.origin)
1155 else:
1156 name = absolute_name
1157 rdtype, rdclass, ttl, rdlen = self.parser.get_struct("!HHIH")
1158 if rdtype in (dns.rdatatype.OPT, dns.rdatatype.TSIG):
1159 (
1160 rdclass,
1161 rdtype,
1162 deleting,
1163 empty,
1164 ) = self.message._parse_special_rr_header(
1165 section_number, count, i, name, rdclass, rdtype
1166 )
1167 else:
1168 rdclass, rdtype, deleting, empty = self.message._parse_rr_header(
1169 section_number, name, rdclass, rdtype
1170 )
1171 rdata_start = self.parser.current
1172 try:
1173 if empty:
1174 if rdlen > 0:
1175 raise dns.exception.FormError
1176 rd = None
1177 covers = dns.rdatatype.NONE
1178 else:
1179 with self.parser.restrict_to(rdlen):
1180 rd = dns.rdata.from_wire_parser(
1181 rdclass, # pyright: ignore
1182 rdtype,
1183 self.parser,
1184 self.message.origin,
1185 )
1186 covers = rd.covers()
1187 if self.message.xfr and rdtype == dns.rdatatype.SOA:
1188 force_unique = True
1189 if rdtype == dns.rdatatype.OPT:
1190 self.message.opt = dns.rrset.from_rdata(name, ttl, rd)
1191 elif rdtype == dns.rdatatype.TSIG:
1192 trd = cast(dns.rdtypes.ANY.TSIG.TSIG, rd)
1193 if self.keyring is None or self.keyring is True:
1194 raise UnknownTSIGKey("got signed message without keyring")
1195 elif isinstance(self.keyring, dict):
1196 key = self.keyring.get(absolute_name)
1197 if isinstance(key, bytes):
1198 key = dns.tsig.Key(absolute_name, key, trd.algorithm)
1199 elif callable(self.keyring):
1200 key = self.keyring(self.message, absolute_name)
1201 else:
1202 key = self.keyring
1203 if key is None:
1204 raise UnknownTSIGKey(f"key '{name}' unknown")
1205 if key:
1206 self.message.keyring = key
1207 self.message.tsig_ctx = dns.tsig.validate(
1208 self.parser.wire,
1209 key,
1210 absolute_name,
1211 rd,
1212 int(time.time()),
1213 self.message.request_mac,
1214 rr_start,
1215 self.message.tsig_ctx,
1216 self.multi,
1217 )
1218 self.message.tsig = dns.rrset.from_rdata(absolute_name, 0, rd)
1219 else:
1220 rrset = self.message.find_rrset(
1221 section,
1222 name,
1223 rdclass, # pyright: ignore
1224 rdtype,
1225 covers,
1226 deleting,
1227 True,
1228 force_unique,
1229 )
1230 if rd is not None:
1231 if ttl > 0x7FFFFFFF:
1232 ttl = 0
1233 rrset.add(rd, ttl)
1234 except Exception as e:
1235 if self.continue_on_error:
1236 self._add_error(e)
1237 self.parser.seek(rdata_start + rdlen)
1238 else:
1239 raise
1241 def read(self):
1242 """Read a wire format DNS message and build a dns.message.Message
1243 object."""
1245 if self.parser.remaining() < 12:
1246 raise ShortHeader
1247 id, flags, qcount, ancount, aucount, adcount = self.parser.get_struct("!HHHHHH")
1248 factory = _message_factory_from_opcode(dns.opcode.from_flags(flags))
1249 self.message = factory(id=id)
1250 self.message.flags = dns.flags.Flag(flags)
1251 self.message.wire = self.parser.wire
1252 self.initialize_message(self.message)
1253 self.one_rr_per_rrset = self.message._get_one_rr_per_rrset(
1254 self.one_rr_per_rrset
1255 )
1256 try:
1257 self._get_question(MessageSection.QUESTION, qcount)
1258 if self.question_only:
1259 return self.message
1260 self._get_section(MessageSection.ANSWER, ancount)
1261 self._get_section(MessageSection.AUTHORITY, aucount)
1262 self._get_section(MessageSection.ADDITIONAL, adcount)
1263 if not self.ignore_trailing and self.parser.remaining() != 0:
1264 raise TrailingJunk
1265 if self.multi:
1266 if self.message.tsig_ctx and not self.message.had_tsig:
1267 # In the (obsolete) multi-mode, and we didn't get a TSIG, then
1268 # we need to update the context with this message.
1269 self.message.tsig_ctx.update(self.parser.wire)
1270 elif (
1271 self.keyring is not False
1272 and self.message.request_mac
1273 and not self.message.had_tsig
1274 ):
1275 # A TSIG is required but not present.
1276 raise dns.exception.FormError(
1277 "response requires a TSIG but does not have one"
1278 )
1279 except Exception as e:
1280 if self.continue_on_error:
1281 self._add_error(e)
1282 else:
1283 raise
1284 return self.message
1287def from_wire(
1288 wire: bytes,
1289 keyring: Any | None = None,
1290 request_mac: bytes | None = b"",
1291 xfr: bool = False,
1292 origin: dns.name.Name | None = None,
1293 tsig_ctx: dns.tsig.HMACTSig | dns.tsig.GSSTSig | None = None,
1294 multi: bool = False,
1295 question_only: bool = False,
1296 one_rr_per_rrset: bool = False,
1297 ignore_trailing: bool = False,
1298 raise_on_truncation: bool = False,
1299 continue_on_error: bool = False,
1300) -> Message:
1301 """Convert a DNS wire format message into a message object.
1303 :param keyring: The key or keyring for TSIG validation. ``None`` or
1304 ``True`` causes TSIG-signed messages to fail; ``False`` disables
1305 validation. When used to validate responses to TSIG-signed requests,
1306 the keyring should only have the key used to sign the request, otherwise
1307 the validation will not correctly check that the key name and algorithm
1308 match the request.
1309 :type keyring: :py:class:`dns.tsig.Key`, dict, bool, or ``None``
1310 :param request_mac: MAC of the TSIG-signed request this message responds
1311 to, if any. If nonempty, the message must have a TSIG unless *multi*
1312 is ``True`` or *keyring* is ``False``.
1313 :type request_mac: bytes or ``None``
1314 :param xfr: ``True`` if this message is part of a zone transfer.
1315 :type xfr: bool
1316 :param origin: Zone origin for zone transfers; names are relativized to
1317 this if not ``None``.
1318 :type origin: :py:class:`dns.name.Name` or ``None``
1319 :param tsig_ctx: Ongoing TSIG context for zone transfer validation.
1320 :param multi: ``True`` if this message is part of a multi-message sequence.
1321 :type multi: bool
1322 :param question_only: If ``True``, read only the question section.
1323 :type question_only: bool
1324 :param one_rr_per_rrset: If ``True``, put each RR into its own RRset.
1325 :type one_rr_per_rrset: bool
1326 :param ignore_trailing: If ``True``, ignore trailing octets after the
1327 message.
1328 :type ignore_trailing: bool
1329 :param raise_on_truncation: If ``True``, raise
1330 :py:exc:`dns.message.Truncated` when the TC bit is set.
1331 :type raise_on_truncation: bool
1332 :param continue_on_error: If ``True``, try to continue parsing on errors
1333 and accumulate them in the message's ``errors`` attribute.
1334 :type continue_on_error: bool
1335 :raises dns.message.ShortHeader: If the message is less than 12 octets.
1336 :raises dns.message.TrailingJunk: If trailing octets are present and
1337 *ignore_trailing* is ``False``.
1338 :raises dns.message.BadEDNS: If an OPT record is in the wrong section.
1339 :raises dns.message.BadTSIG: If a TSIG record is not the last additional
1340 record.
1341 :raises dns.message.Truncated: If the TC flag is set and
1342 *raise_on_truncation* is ``True``.
1343 :raises dns.exception.FormError: If *request_mac* is nonempty, *multi* is
1344 ``False``, *keyring* is not ``False``, and the message does not have a
1345 TSIG.
1346 :rtype: :py:class:`dns.message.Message`
1347 """
1349 # We permit None for request_mac solely for backwards compatibility
1350 if request_mac is None:
1351 request_mac = b""
1353 def initialize_message(message):
1354 message.request_mac = request_mac
1355 message.xfr = xfr
1356 message.origin = origin
1357 message.tsig_ctx = tsig_ctx
1359 reader = _WireReader(
1360 wire,
1361 initialize_message,
1362 question_only,
1363 one_rr_per_rrset,
1364 ignore_trailing,
1365 keyring,
1366 multi,
1367 continue_on_error,
1368 )
1369 try:
1370 m = reader.read()
1371 except dns.exception.FormError:
1372 if (
1373 reader.message
1374 and (reader.message.flags & dns.flags.TC)
1375 and raise_on_truncation
1376 ):
1377 raise Truncated(message=reader.message)
1378 else:
1379 raise
1380 # Reading a truncated message might not have any errors, so we
1381 # have to do this check here too.
1382 if m.flags & dns.flags.TC and raise_on_truncation:
1383 raise Truncated(message=m)
1384 if continue_on_error:
1385 m.errors = reader.errors
1387 return m
1390class _TextReader:
1391 """Text format reader.
1393 tok: the tokenizer.
1394 message: The message object being built.
1395 DNS dynamic updates.
1396 last_name: The most recently read name when building a message object.
1397 one_rr_per_rrset: Put each RR into its own RRset?
1398 origin: The origin for relative names
1399 relativize: relativize names?
1400 relativize_to: the origin to relativize to.
1401 """
1403 def __init__(
1404 self,
1405 text: Any,
1406 idna_codec: dns.name.IDNACodec | None,
1407 one_rr_per_rrset: bool = False,
1408 origin: dns.name.Name | None = None,
1409 relativize: bool = True,
1410 relativize_to: dns.name.Name | None = None,
1411 ):
1412 self.message: Message | None = None # mypy: ignore
1413 self.tok = dns.tokenizer.Tokenizer(text, idna_codec=idna_codec)
1414 self.last_name = None
1415 self.one_rr_per_rrset = one_rr_per_rrset
1416 self.origin = origin
1417 self.relativize = relativize
1418 self.relativize_to = relativize_to
1419 self.id = None
1420 self.edns = -1
1421 self.ednsflags = 0
1422 self.payload = DEFAULT_EDNS_PAYLOAD
1423 self.rcode = None
1424 self.opcode = dns.opcode.QUERY
1425 self.flags = 0
1427 def _header_line(self, _):
1428 """Process one line from the text format header section."""
1430 token = self.tok.get()
1431 what = token.value
1432 if what == "id":
1433 self.id = self.tok.get_int()
1434 elif what == "flags":
1435 while True:
1436 token = self.tok.get()
1437 if not token.is_identifier():
1438 self.tok.unget(token)
1439 break
1440 self.flags = self.flags | dns.flags.from_text(token.value)
1441 elif what == "edns":
1442 self.edns = self.tok.get_int()
1443 self.ednsflags = self.ednsflags | (self.edns << 16)
1444 elif what == "eflags":
1445 if self.edns < 0:
1446 self.edns = 0
1447 while True:
1448 token = self.tok.get()
1449 if not token.is_identifier():
1450 self.tok.unget(token)
1451 break
1452 self.ednsflags = self.ednsflags | dns.flags.edns_from_text(token.value)
1453 elif what == "payload":
1454 self.payload = self.tok.get_int()
1455 if self.edns < 0:
1456 self.edns = 0
1457 elif what == "opcode":
1458 text = self.tok.get_string()
1459 self.opcode = dns.opcode.from_text(text)
1460 self.flags = self.flags | dns.opcode.to_flags(self.opcode)
1461 elif what == "rcode":
1462 text = self.tok.get_string()
1463 self.rcode = dns.rcode.from_text(text)
1464 else:
1465 raise UnknownHeaderField
1466 self.tok.get_eol()
1468 def _question_line(self, section_number):
1469 """Process one line from the text format question section."""
1471 assert self.message is not None
1472 section = self.message.sections[section_number]
1473 token = self.tok.get(want_leading=True)
1474 if not token.is_whitespace():
1475 self.last_name = self.tok.as_name(
1476 token, self.message.origin, self.relativize, self.relativize_to
1477 )
1478 name = self.last_name
1479 if name is None:
1480 raise NoPreviousName
1481 token = self.tok.get()
1482 if not token.is_identifier():
1483 raise dns.exception.SyntaxError
1484 # Class
1485 try:
1486 rdclass = dns.rdataclass.from_text(token.value)
1487 token = self.tok.get()
1488 if not token.is_identifier():
1489 raise dns.exception.SyntaxError
1490 except dns.exception.SyntaxError:
1491 raise dns.exception.SyntaxError
1492 except Exception:
1493 rdclass = dns.rdataclass.IN
1494 # Type
1495 rdtype = dns.rdatatype.from_text(token.value)
1496 rdclass, rdtype, _, _ = self.message._parse_rr_header(
1497 section_number, name, rdclass, rdtype
1498 )
1499 self.message.find_rrset(
1500 section, name, rdclass, rdtype, create=True, force_unique=True
1501 )
1502 self.tok.get_eol()
1504 def _rr_line(self, section_number):
1505 """Process one line from the text format answer, authority, or
1506 additional data sections.
1507 """
1509 assert self.message is not None
1510 section = self.message.sections[section_number]
1511 # Name
1512 token = self.tok.get(want_leading=True)
1513 if not token.is_whitespace():
1514 self.last_name = self.tok.as_name(
1515 token, self.message.origin, self.relativize, self.relativize_to
1516 )
1517 name = self.last_name
1518 if name is None:
1519 raise NoPreviousName
1520 token = self.tok.get()
1521 if not token.is_identifier():
1522 raise dns.exception.SyntaxError
1523 # TTL
1524 try:
1525 ttl = dns.ttl.from_text(token.value)
1526 except dns.ttl.BadTTL:
1527 print("bad ttl")
1528 ttl = 0
1529 else:
1530 token = self.tok.get()
1531 if not token.is_identifier():
1532 raise dns.exception.SyntaxError
1533 # Class
1534 try:
1535 rdclass = dns.rdataclass.from_text(token.value)
1536 token = self.tok.get()
1537 if not token.is_identifier():
1538 raise dns.exception.SyntaxError
1539 except dns.exception.SyntaxError:
1540 raise dns.exception.SyntaxError
1541 except Exception:
1542 rdclass = dns.rdataclass.IN
1543 # Type
1544 rdtype = dns.rdatatype.from_text(token.value)
1545 rdclass, rdtype, deleting, empty = self.message._parse_rr_header(
1546 section_number, name, rdclass, rdtype
1547 )
1548 token = self.tok.get()
1549 if empty and not token.is_eol_or_eof():
1550 raise dns.exception.SyntaxError
1551 if not empty and token.is_eol_or_eof():
1552 raise dns.exception.UnexpectedEnd
1553 if not token.is_eol_or_eof():
1554 self.tok.unget(token)
1555 rd = dns.rdata.from_text(
1556 rdclass,
1557 rdtype,
1558 self.tok,
1559 self.message.origin,
1560 self.relativize,
1561 self.relativize_to,
1562 )
1563 covers = rd.covers()
1564 else:
1565 rd = None
1566 covers = dns.rdatatype.NONE
1567 rrset = self.message.find_rrset(
1568 section,
1569 name,
1570 rdclass,
1571 rdtype,
1572 covers,
1573 deleting,
1574 True,
1575 self.one_rr_per_rrset,
1576 )
1577 if rd is not None:
1578 rrset.add(rd, ttl)
1580 def _make_message(self):
1581 factory = _message_factory_from_opcode(self.opcode)
1582 message = factory(id=self.id)
1583 message.flags = self.flags
1584 if self.edns >= 0:
1585 message.use_edns(self.edns, self.ednsflags, self.payload)
1586 if self.rcode:
1587 message.set_rcode(self.rcode)
1588 if self.origin:
1589 message.origin = self.origin
1590 return message
1592 def read(self):
1593 """Read a text format DNS message and build a dns.message.Message
1594 object."""
1596 line_method = self._header_line
1597 section_number = None
1598 while 1:
1599 token = self.tok.get(True, True)
1600 if token.is_eol_or_eof():
1601 break
1602 if token.is_comment():
1603 u = token.value.upper()
1604 if u == "HEADER":
1605 line_method = self._header_line
1607 if self.message:
1608 message = self.message
1609 else:
1610 # If we don't have a message, create one with the current
1611 # opcode, so that we know which section names to parse.
1612 message = self._make_message()
1613 try:
1614 section_number = message._section_enum.from_text(u)
1615 # We found a section name. If we don't have a message,
1616 # use the one we just created.
1617 if not self.message:
1618 self.message = message
1619 self.one_rr_per_rrset = message._get_one_rr_per_rrset(
1620 self.one_rr_per_rrset
1621 )
1622 if section_number == MessageSection.QUESTION:
1623 line_method = self._question_line
1624 else:
1625 line_method = self._rr_line
1626 except Exception:
1627 # It's just a comment.
1628 pass
1629 self.tok.get_eol()
1630 continue
1631 self.tok.unget(token)
1632 line_method(section_number)
1633 if not self.message:
1634 self.message = self._make_message()
1635 return self.message
1638def from_text(
1639 text: str,
1640 idna_codec: dns.name.IDNACodec | None = None,
1641 one_rr_per_rrset: bool = False,
1642 origin: dns.name.Name | None = None,
1643 relativize: bool = True,
1644 relativize_to: dns.name.Name | None = None,
1645) -> Message:
1646 """Convert the text format message into a message object.
1648 The reader stops after reading the first blank line in the input to
1649 facilitate reading multiple messages from a single file with
1650 ``dns.message.from_file()``.
1652 :param text: The text format message.
1653 :type text: str
1654 :param idna_codec: The IDNA encoder/decoder. Defaults to IDNA 2003.
1655 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
1656 :param one_rr_per_rrset: If ``True``, put each RR into its own RRset.
1657 :type one_rr_per_rrset: bool
1658 :param origin: The origin to use for relative names.
1659 :type origin: :py:class:`dns.name.Name` or ``None``
1660 :param relativize: If ``True``, names will be relativized.
1661 :type relativize: bool
1662 :param relativize_to: The origin to relativize to. Defaults to *origin*.
1663 :type relativize_to: :py:class:`dns.name.Name` or ``None``
1664 :raises dns.message.UnknownHeaderField: If a header field is unknown.
1665 :raises dns.exception.SyntaxError: If the text is badly formed.
1666 :rtype: :py:class:`dns.message.Message`
1667 """
1669 # 'text' can also be a file, but we don't publish that fact
1670 # since it's an implementation detail. The official file
1671 # interface is from_file().
1673 reader = _TextReader(
1674 text, idna_codec, one_rr_per_rrset, origin, relativize, relativize_to
1675 )
1676 return reader.read()
1679def from_file(
1680 f: str | os.PathLike | TextIO,
1681 idna_codec: dns.name.IDNACodec | None = None,
1682 one_rr_per_rrset: bool = False,
1683) -> Message:
1684 """Read the next text format message from the specified file.
1686 Message blocks are separated by a single blank line.
1688 :param f: A file object, or the name of a file to open, as a ``str``
1689 or an ``os.PathLike``.
1690 :param idna_codec: The IDNA encoder/decoder. Defaults to IDNA 2003.
1691 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
1692 :param one_rr_per_rrset: If ``True``, put each RR into its own RRset.
1693 :type one_rr_per_rrset: bool
1694 :raises dns.message.UnknownHeaderField: If a header field is unknown.
1695 :raises dns.exception.SyntaxError: If the text is badly formed.
1696 :rtype: :py:class:`dns.message.Message`
1697 """
1699 with dns._file_util.maybe_open(f, encoding="utf-8") as fp:
1700 reader = _TextReader(fp, idna_codec, one_rr_per_rrset)
1701 return reader.read()
1702 assert False # for mypy lgtm[py/unreachable-statement]
1705def make_query(
1706 qname: dns.name.Name | str,
1707 rdtype: dns.rdatatype.RdataType | str,
1708 rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
1709 use_edns: int | bool | None = None,
1710 want_dnssec: bool = False,
1711 ednsflags: int | None = None,
1712 payload: int | None = None,
1713 request_payload: int | None = None,
1714 options: list[dns.edns.Option] | None = None,
1715 idna_codec: dns.name.IDNACodec | None = None,
1716 id: int | None = None,
1717 flags: int = dns.flags.RD,
1718 pad: int = 0,
1719) -> QueryMessage:
1720 """Make a query message.
1722 The query name, type, and class may be specified as objects of the
1723 appropriate type or as strings. The query id is chosen at random, and
1724 the DNS flags are set to ``dns.flags.RD``.
1726 :param qname: The query name.
1727 :type qname: :py:class:`dns.name.Name` or ``str``
1728 :param rdtype: The desired rdata type.
1729 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
1730 :param rdclass: The desired rdata class; default is IN.
1731 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
1732 :param use_edns: The EDNS level; ``None`` enables EDNS only if other EDNS
1733 parameters are set. See :py:meth:`dns.message.Message.use_edns`.
1734 :type use_edns: int, bool, or ``None``
1735 :param want_dnssec: If ``True``, DNSSEC data is desired.
1736 :type want_dnssec: bool
1737 :param ednsflags: The EDNS flag values.
1738 :type ednsflags: int
1739 :param payload: The EDNS sender payload field (max UDP response size).
1740 :type payload: int
1741 :param request_payload: The EDNS payload size to advertise when sending.
1742 Defaults to *payload*.
1743 :type request_payload: int
1744 :param options: The EDNS options.
1745 :type options: list of :py:class:`dns.edns.Option` or ``None``
1746 :param idna_codec: The IDNA encoder/decoder. Defaults to IDNA 2003.
1747 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
1748 :param id: The query id. Defaults to a random id.
1749 :type id: int or ``None``
1750 :param flags: The query flags. Default is ``dns.flags.RD``.
1751 :type flags: int
1752 :param pad: If non-zero, add EDNS PADDING to make size a multiple of this.
1753 :type pad: int
1754 :rtype: :py:class:`dns.message.QueryMessage`
1755 """
1757 if isinstance(qname, str):
1758 qname = dns.name.from_text(qname, idna_codec=idna_codec)
1759 rdtype = dns.rdatatype.RdataType.make(rdtype)
1760 rdclass = dns.rdataclass.RdataClass.make(rdclass)
1761 m = QueryMessage(id=id)
1762 m.flags = dns.flags.Flag(flags)
1763 m.find_rrset(m.question, qname, rdclass, rdtype, create=True, force_unique=True)
1764 # only pass keywords on to use_edns if they have been set to a
1765 # non-None value. Setting a field will turn EDNS on if it hasn't
1766 # been configured.
1767 kwargs: dict[str, Any] = {}
1768 if ednsflags is not None:
1769 kwargs["ednsflags"] = ednsflags
1770 if payload is not None:
1771 kwargs["payload"] = payload
1772 if request_payload is not None:
1773 kwargs["request_payload"] = request_payload
1774 if options is not None:
1775 kwargs["options"] = options
1776 if kwargs and use_edns is None:
1777 use_edns = 0
1778 kwargs["edns"] = use_edns
1779 kwargs["pad"] = pad
1780 m.use_edns(**kwargs)
1781 if want_dnssec:
1782 m.want_dnssec(want_dnssec)
1783 return m
1786class CopyMode(enum.Enum):
1787 """
1788 How should sections be copied when making an update response?
1789 """
1791 #: Copy no sections; only suitable for testing
1792 NOTHING = 0
1793 #: Copy only the question section, if present.
1794 QUESTION = 1
1795 #: Copy all sections, other than OPT and TSIG RRs.
1796 EVERYTHING = 2
1799def make_response(
1800 query: Message,
1801 recursion_available: bool = False,
1802 our_payload: int = 8192,
1803 fudge: int = 300,
1804 tsig_error: int = 0,
1805 pad: int | None = None,
1806 copy_mode: CopyMode | None = None,
1807) -> Message:
1808 """Make a response skeleton for the specified query.
1810 The returned message has all required response infrastructure but no
1811 content. Copied sections are shallow copies, so the query's RRsets
1812 should not be changed.
1814 :param query: The query to respond to.
1815 :type query: :py:class:`dns.message.Message`
1816 :param recursion_available: If ``True``, set the RA bit.
1817 :type recursion_available: bool
1818 :param our_payload: The EDNS payload size to advertise.
1819 :type our_payload: int
1820 :param fudge: The TSIG time fudge.
1821 :type fudge: int
1822 :param tsig_error: The TSIG error code.
1823 :type tsig_error: int
1824 :param pad: If 0, no padding; if not ``None`` pad to a multiple of this
1825 value; if ``None``, follow RFC 8467 (pad to 468 if request was padded).
1826 :type pad: int or ``None``
1827 :param copy_mode: Controls which sections are copied. ``None`` uses the
1828 default for the opcode (currently
1829 :py:attr:`dns.message.CopyMode.QUESTION`).
1830 :type copy_mode: :py:class:`dns.message.CopyMode` or ``None``
1831 :returns: A response message of the same class as *query*.
1832 :rtype: :py:class:`dns.message.Message`
1833 """
1835 if query.flags & dns.flags.QR:
1836 raise dns.exception.FormError("specified query message is not a query")
1837 opcode = query.opcode()
1838 factory = _message_factory_from_opcode(opcode)
1839 response = factory(id=query.id)
1840 response.flags = dns.flags.QR | (query.flags & dns.flags.RD)
1841 if recursion_available:
1842 response.flags |= dns.flags.RA
1843 response.set_opcode(opcode)
1844 if copy_mode is None:
1845 copy_mode = CopyMode.QUESTION
1846 if copy_mode != CopyMode.NOTHING:
1847 response.question = list(query.question)
1848 if copy_mode == CopyMode.EVERYTHING:
1849 response.answer = list(query.answer)
1850 response.authority = list(query.authority)
1851 response.additional = list(query.additional)
1852 if query.edns >= 0:
1853 if pad is None:
1854 # Set response padding per RFC 8467
1855 pad = 0
1856 for option in query.options:
1857 if option.otype == dns.edns.OptionType.PADDING:
1858 pad = 468
1859 response.use_edns(0, 0, our_payload, query.payload, pad=pad)
1860 if query.had_tsig and query.keyring:
1861 assert query.mac is not None
1862 assert query.keyalgorithm is not None
1863 response.use_tsig(
1864 query.keyring,
1865 query.keyname,
1866 fudge,
1867 None,
1868 tsig_error,
1869 b"",
1870 query.keyalgorithm,
1871 )
1872 response.request_mac = query.mac
1873 return response
1876### BEGIN generated MessageSection constants
1878QUESTION = MessageSection.QUESTION
1879ANSWER = MessageSection.ANSWER
1880AUTHORITY = MessageSection.AUTHORITY
1881ADDITIONAL = MessageSection.ADDITIONAL
1883### END generated MessageSection constants