Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/rdata.py: 28%
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 rdata."""
20import base64
21import binascii
22import dataclasses
23import inspect
24import io
25import ipaddress
26import itertools
27import random
28from importlib import import_module
29from typing import Any
31import dns.exception
32import dns.immutable
33import dns.ipv4
34import dns.ipv6
35import dns.name
36import dns.rdataclass
37import dns.rdatatype
38import dns.tokenizer
39import dns.ttl
40import dns.wire
42_chunksize = 32
44# We currently allow comparisons for rdata with relative names for backwards
45# compatibility, but in the future we will not, as these kinds of comparisons
46# can lead to subtle bugs if code is not carefully written.
47#
48# This switch allows the future behavior to be turned on so code can be
49# tested with it.
50_allow_relative_comparisons = True
53class NoRelativeRdataOrdering(dns.exception.DNSException):
54 """An attempt was made to do an ordered comparison of one or more
55 rdata with relative names. The only reliable way of sorting rdata
56 is to use non-relativized rdata.
58 """
61@dataclasses.dataclass(frozen=True)
62class RdataStyle(dns.name.NameStyle):
63 """Rdata text styles
65 An ``RdataStyle`` is also a :py:class:`dns.name.NameStyle`; see that class
66 for a description of its options.
68 .. attribute:: txt_is_utf8
70 A ``bool``. If ``True``, TXT-like records will be treated as UTF-8 if
71 they decode successfully, and the output string may contain any Unicode
72 codepoint. If ``False`` (the default), TXT-like records are treated
73 according to RFC 1035 rules.
75 .. attribute:: base64_chunk_size
77 An ``int`` (default 32). The chunk size for text representations that
78 break base64 strings into chunks.
80 .. attribute:: base64_chunk_separator
82 A ``str`` (default ``" "``). The separator for text representations
83 that break base64 strings into chunks.
85 .. attribute:: hex_chunk_size
87 An ``int`` (default 128). The chunk size for text representations that
88 break hex strings into chunks.
90 .. attribute:: hex_chunk_separator
92 A ``str`` (default ``" "``). The separator for text representations
93 that break hex strings into chunks.
95 .. attribute:: truncate_crypto
97 A ``bool`` (default ``False``). If ``True``, output of crypto types
98 (e.g. DNSKEY) is altered to be human-readable in a debugging context,
99 but the crypto content is removed. Use of this option will lose
100 information.
101 """
103 txt_is_utf8: bool = False
104 base64_chunk_size: int = 32
105 base64_chunk_separator: str = " "
106 hex_chunk_size: int = 128
107 hex_chunk_separator: str = " "
108 truncate_crypto: bool = False
111def _wordbreak(
112 data: bytes, chunksize: int = _chunksize, separator: bytes | str = b" "
113) -> str:
114 """Break a binary string into chunks of chunksize characters separated by
115 a space.
116 """
118 if isinstance(separator, str):
119 separator = separator.encode()
120 if not chunksize:
121 return data.decode()
122 return separator.join(
123 [data[i : i + chunksize] for i in range(0, len(data), chunksize)]
124 ).decode()
127# pylint: disable=unused-argument
130def _hexify(data, chunksize=_chunksize, separator: bytes | str = " ", **kw):
131 """Convert a binary string into its hex encoding, broken up into chunks
132 of chunksize characters separated by a separator.
133 """
134 if isinstance(separator, bytes):
135 separator = separator.decode()
136 return _styled_hexify(
137 data, RdataStyle(hex_chunk_separator=separator, hex_chunk_size=chunksize)
138 )
141def _styled_hexify(data, style: RdataStyle, is_crypto: bool = False):
142 """Convert a binary string into its hex encoding, broken up into chunks
143 of characters separated by a separator.
144 """
146 if style.truncate_crypto and is_crypto:
147 return "[omitted]"
148 return _wordbreak(
149 binascii.hexlify(data),
150 style.hex_chunk_size,
151 style.hex_chunk_separator,
152 )
155def _base64ify(data, chunksize=_chunksize, separator: bytes | str = " ", **kw):
156 """Convert a binary string into its base64 encoding, broken up into chunks
157 of chunksize characters separated by a separator.
158 """
159 if isinstance(separator, bytes):
160 separator = separator.decode()
161 return _styled_base64ify(
162 data, RdataStyle(base64_chunk_separator=separator, base64_chunk_size=chunksize)
163 )
166def _styled_base64ify(data, style: RdataStyle, is_crypto: bool = False):
167 """Convert a binary string into its base64 encoding, broken up into chunks
168 of characters separated by a separator.
169 """
171 if style.truncate_crypto and is_crypto:
172 return "[omitted]"
173 return _wordbreak(
174 base64.b64encode(data),
175 style.base64_chunk_size,
176 style.base64_chunk_separator,
177 )
180# pylint: enable=unused-argument
182_escaped = b'"\\'
183_unicode_escaped = '"\\'
186def _escapify(qstring):
187 """Escape the characters in a quoted string which need it."""
189 if isinstance(qstring, str):
190 qstring = qstring.encode()
191 if not isinstance(qstring, bytearray):
192 qstring = bytearray(qstring)
194 text = ""
195 for c in qstring:
196 if c in _escaped:
197 text += "\\" + chr(c)
198 elif c >= 0x20 and c < 0x7F:
199 text += chr(c)
200 else:
201 text += f"\\{c:03d}"
202 return text
205def _escapify_unicode(qstring):
206 """Escape the characters in a Unicode quoted string which need it."""
208 text = ""
209 for c in qstring:
210 if c in _unicode_escaped:
211 text += "\\" + c
212 elif ord(c) >= 0x20:
213 text += c
214 else:
215 text += f"\\{ord(c):03d}"
216 return text
219def _truncate_bitmap(what):
220 """Determine the index of greatest byte that isn't all zeros, and
221 return the bitmap that contains all the bytes less than that index.
222 """
224 for i in range(len(what) - 1, -1, -1):
225 if what[i] != 0:
226 return what[0 : i + 1]
227 return what[0:1]
230# So we don't have to edit all the rdata classes...
231_constify = dns.immutable.constify
234@dns.immutable.immutable
235class Rdata:
236 """Base class for all DNS rdata types."""
238 __slots__ = ["rdclass", "rdtype", "rdcomment"]
240 _crypto_keep_first_n: int | None = None
242 def __init__(
243 self,
244 rdclass: dns.rdataclass.RdataClass,
245 rdtype: dns.rdatatype.RdataType,
246 ) -> None:
247 """Initialize an rdata.
249 :param rdclass: The rdata class.
250 :type rdclass: :py:class:`dns.rdataclass.RdataClass`
251 :param rdtype: The rdata type.
252 :type rdtype: :py:class:`dns.rdatatype.RdataType`
253 """
255 self.rdclass = self._as_rdataclass(rdclass)
256 self.rdtype = self._as_rdatatype(rdtype)
257 self.rdcomment = None
259 def _get_all_slots(self):
260 return itertools.chain.from_iterable(
261 getattr(cls, "__slots__", []) for cls in self.__class__.__mro__
262 )
264 def __getstate__(self):
265 # We used to try to do a tuple of all slots here, but it
266 # doesn't work as self._all_slots isn't available at
267 # __setstate__() time. Before that we tried to store a tuple
268 # of __slots__, but that didn't work as it didn't store the
269 # slots defined by ancestors. This older way didn't fail
270 # outright, but ended up with partially broken objects, e.g.
271 # if you unpickled an A RR it wouldn't have rdclass and rdtype
272 # attributes, and would compare badly.
273 state = {}
274 for slot in self._get_all_slots():
275 state[slot] = getattr(self, slot)
276 return state
278 def __setstate__(self, state):
279 for slot, val in state.items():
280 object.__setattr__(self, slot, val)
281 if not hasattr(self, "rdcomment"):
282 # Pickled rdata from 2.0.x might not have a rdcomment, so add
283 # it if needed.
284 object.__setattr__(self, "rdcomment", None)
286 def covers(self) -> dns.rdatatype.RdataType:
287 """Return the type a Rdata covers.
289 DNS SIG/RRSIG rdatas apply to a specific type; this type is
290 returned by the covers() function. If the rdata type is not
291 SIG or RRSIG, dns.rdatatype.NONE is returned. This is useful when
292 creating rdatasets, allowing the rdataset to contain only RRSIGs
293 of a particular type, e.g. RRSIG(NS).
295 :rtype: :py:class:`dns.rdatatype.RdataType`
296 """
298 return dns.rdatatype.NONE
300 def extended_rdatatype(self) -> int:
301 """Return a 32-bit type value, the least significant 16 bits of
302 which are the ordinary DNS type, and the upper 16 bits of which are
303 the "covered" type, if any.
305 :rtype: int
306 """
308 return self.covers() << 16 | self.rdtype
310 def to_text(
311 self,
312 origin: dns.name.Name | None = None,
313 relativize: bool = True,
314 style: RdataStyle | None = None,
315 **kw: Any,
316 ) -> str:
317 """Convert an rdata to text format.
319 :param style: If specified, overrides *origin* and *relativize*.
320 :type style: :py:class:`dns.rdata.RdataStyle` or ``None``
321 :rtype: str
322 """
323 if style is None:
324 kw = kw.copy()
325 kw["origin"] = origin
326 kw["relativize"] = relativize
327 style = RdataStyle.from_keywords(kw)
328 return self.to_styled_text(style)
330 def to_styled_text(self, style: RdataStyle) -> str:
331 """Convert an rdata to styled text format.
333 See the documentation for :py:class:`dns.rdata.RdataStyle` for a description
334 of the style parameters.
336 :rtype: str
337 """
339 raise NotImplementedError # pragma: no cover
341 def _to_wire(
342 self,
343 file: Any,
344 compress: dns.name.CompressType | None = None,
345 origin: dns.name.Name | None = None,
346 canonicalize: bool = False,
347 ) -> None:
348 raise NotImplementedError # pragma: no cover
350 def to_wire(
351 self,
352 file: Any | None = None,
353 compress: dns.name.CompressType | None = None,
354 origin: dns.name.Name | None = None,
355 canonicalize: bool = False,
356 ) -> bytes | None:
357 """Convert an rdata to wire format.
359 :returns: Wire-format bytes if no output file was specified, or ``None``
360 if a file was provided.
361 :rtype: bytes or ``None``
362 """
364 if file:
365 # We call _to_wire() and then return None explicitly instead of
366 # of just returning the None from _to_wire() as mypy's func-returns-value
367 # unhelpfully errors out with "error: "_to_wire" of "Rdata" does not return
368 # a value (it only ever returns None)"
369 self._to_wire(file, compress, origin, canonicalize)
370 return None
371 else:
372 f = io.BytesIO()
373 self._to_wire(f, compress, origin, canonicalize)
374 return f.getvalue()
376 def to_generic(self, origin: dns.name.Name | None = None) -> "GenericRdata":
377 """Create a :py:class:`dns.rdata.GenericRdata` equivalent of this rdata.
379 :rtype: :py:class:`dns.rdata.GenericRdata`
380 """
381 wire = self.to_wire(origin=origin)
382 assert wire is not None # for type checkers
383 return GenericRdata(self.rdclass, self.rdtype, wire)
385 def to_digestable(self, origin: dns.name.Name | None = None) -> bytes:
386 """Convert rdata to a format suitable for digesting in hashes. This
387 is also the DNSSEC canonical form.
389 :rtype: bytes
390 """
391 wire = self.to_wire(origin=origin, canonicalize=True)
392 assert wire is not None # for mypy
393 return wire
395 def __repr__(self):
396 covers = self.covers()
397 if covers == dns.rdatatype.NONE:
398 ctext = ""
399 else:
400 ctext = "(" + dns.rdatatype.to_text(covers) + ")"
401 return (
402 "<DNS "
403 + dns.rdataclass.to_text(self.rdclass)
404 + " "
405 + dns.rdatatype.to_text(self.rdtype)
406 + ctext
407 + " rdata: "
408 + str(self)
409 + ">"
410 )
412 def __str__(self):
413 return self.to_text()
415 def _cmp(self, other):
416 """Compare an rdata with another rdata of the same rdtype and
417 rdclass.
419 For rdata with only absolute names:
420 Return < 0 if self < other in the DNSSEC ordering, 0 if self
421 == other, and > 0 if self > other.
422 For rdata with at least one relative names:
423 The rdata sorts before any rdata with only absolute names.
424 When compared with another relative rdata, all names are
425 made absolute as if they were relative to the root, as the
426 proper origin is not available. While this creates a stable
427 ordering, it is NOT guaranteed to be the DNSSEC ordering.
428 In the future, all ordering comparisons for rdata with
429 relative names will be disallowed.
430 """
431 # the next two lines are for type checkers, so they are bound
432 our = b""
433 their = b""
434 try:
435 our = self.to_digestable()
436 our_relative = False
437 except dns.name.NeedAbsoluteNameOrOrigin:
438 if _allow_relative_comparisons:
439 our = self.to_digestable(dns.name.root)
440 our_relative = True
441 try:
442 their = other.to_digestable()
443 their_relative = False
444 except dns.name.NeedAbsoluteNameOrOrigin:
445 if _allow_relative_comparisons:
446 their = other.to_digestable(dns.name.root)
447 their_relative = True
448 if _allow_relative_comparisons:
449 if our_relative != their_relative:
450 # For the purpose of comparison, all rdata with at least one
451 # relative name is less than an rdata with only absolute names.
452 if our_relative:
453 return -1
454 else:
455 return 1
456 elif our_relative or their_relative:
457 raise NoRelativeRdataOrdering
458 if our == their:
459 return 0
460 elif our > their:
461 return 1
462 else:
463 return -1
465 def __eq__(self, other):
466 if not isinstance(other, Rdata):
467 return False
468 if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
469 return False
470 our_relative = False
471 their_relative = False
472 try:
473 our = self.to_digestable()
474 except dns.name.NeedAbsoluteNameOrOrigin:
475 our = self.to_digestable(dns.name.root)
476 our_relative = True
477 try:
478 their = other.to_digestable()
479 except dns.name.NeedAbsoluteNameOrOrigin:
480 their = other.to_digestable(dns.name.root)
481 their_relative = True
482 if our_relative != their_relative:
483 return False
484 return our == their
486 def __ne__(self, other):
487 if not isinstance(other, Rdata):
488 return True
489 if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
490 return True
491 return not self.__eq__(other)
493 def __lt__(self, other):
494 if (
495 not isinstance(other, Rdata)
496 or self.rdclass != other.rdclass
497 or self.rdtype != other.rdtype
498 ):
499 return NotImplemented
500 return self._cmp(other) < 0
502 def __le__(self, other):
503 if (
504 not isinstance(other, Rdata)
505 or self.rdclass != other.rdclass
506 or self.rdtype != other.rdtype
507 ):
508 return NotImplemented
509 return self._cmp(other) <= 0
511 def __ge__(self, other):
512 if (
513 not isinstance(other, Rdata)
514 or self.rdclass != other.rdclass
515 or self.rdtype != other.rdtype
516 ):
517 return NotImplemented
518 return self._cmp(other) >= 0
520 def __gt__(self, other):
521 if (
522 not isinstance(other, Rdata)
523 or self.rdclass != other.rdclass
524 or self.rdtype != other.rdtype
525 ):
526 return NotImplemented
527 return self._cmp(other) > 0
529 def __hash__(self):
530 return hash(self.to_digestable(dns.name.root))
532 @classmethod
533 def from_text(
534 cls,
535 rdclass: dns.rdataclass.RdataClass,
536 rdtype: dns.rdatatype.RdataType,
537 tok: dns.tokenizer.Tokenizer,
538 origin: dns.name.Name | None = None,
539 relativize: bool = True,
540 relativize_to: dns.name.Name | None = None,
541 ) -> "Rdata":
542 raise NotImplementedError # pragma: no cover
544 @classmethod
545 def from_wire_parser(
546 cls,
547 rdclass: dns.rdataclass.RdataClass,
548 rdtype: dns.rdatatype.RdataType,
549 parser: dns.wire.Parser,
550 origin: dns.name.Name | None = None,
551 ) -> "Rdata":
552 raise NotImplementedError # pragma: no cover
554 def replace(self, **kwargs: Any) -> "Rdata":
555 """Create a new Rdata instance based on the instance replace was
556 invoked on. It is possible to pass different parameters to
557 override the corresponding properties of the base Rdata.
559 Any field specific to the Rdata type can be replaced, but the
560 *rdtype* and *rdclass* fields cannot.
562 :returns: A new instance of the same :py:class:`Rdata` subclass as *self*.
563 :rtype: :py:class:`Rdata`
564 """
566 # Get the constructor parameters.
567 parameters = inspect.signature(self.__init__).parameters # pyright: ignore
569 # Ensure that all of the arguments correspond to valid fields.
570 # Don't allow rdclass or rdtype to be changed, though.
571 for key in kwargs:
572 if key == "rdcomment":
573 continue
574 if key not in parameters:
575 raise AttributeError(
576 f"'{self.__class__.__name__}' object has no attribute '{key}'"
577 )
578 if key in ("rdclass", "rdtype"):
579 raise AttributeError(
580 f"Cannot overwrite '{self.__class__.__name__}' attribute '{key}'"
581 )
583 # Construct the parameter list. For each field, use the value in
584 # kwargs if present, and the current value otherwise.
585 args = (kwargs.get(key, getattr(self, key)) for key in parameters)
587 # Create, validate, and return the new object.
588 rd = self.__class__(*args)
589 # The comment is not set in the constructor, so give it special
590 # handling.
591 rdcomment = kwargs.get("rdcomment", self.rdcomment)
592 if rdcomment is not None:
593 object.__setattr__(rd, "rdcomment", rdcomment)
594 return rd
596 # Type checking and conversion helpers. These are class methods as
597 # they don't touch object state and may be useful to others.
599 @classmethod
600 def _as_rdataclass(cls, value):
601 return dns.rdataclass.RdataClass.make(value)
603 @classmethod
604 def _as_rdatatype(cls, value):
605 return dns.rdatatype.RdataType.make(value)
607 @classmethod
608 def _as_bytes(
609 cls,
610 value: Any,
611 encode: bool = False,
612 max_length: int | None = None,
613 empty_ok: bool = True,
614 ) -> bytes:
615 if encode and isinstance(value, str):
616 bvalue = value.encode()
617 elif isinstance(value, bytearray):
618 bvalue = bytes(value)
619 elif isinstance(value, bytes):
620 bvalue = value
621 else:
622 raise ValueError("not bytes")
623 if max_length is not None and len(bvalue) > max_length:
624 raise ValueError("too long")
625 if not empty_ok and len(bvalue) == 0:
626 raise ValueError("empty bytes not allowed")
627 return bvalue
629 @classmethod
630 def _as_name(cls, value):
631 # Note that proper name conversion (e.g. with origin and IDNA
632 # awareness) is expected to be done via from_text. This is just
633 # a simple thing for people invoking the constructor directly.
634 if isinstance(value, str):
635 return dns.name.from_text(value)
636 elif not isinstance(value, dns.name.Name):
637 raise ValueError("not a name")
638 return value
640 @classmethod
641 def _as_uint8(cls, value):
642 if not isinstance(value, int):
643 raise ValueError("not an integer")
644 if value < 0 or value > 255:
645 raise ValueError("not a uint8")
646 return value
648 @classmethod
649 def _as_uint16(cls, value):
650 if not isinstance(value, int):
651 raise ValueError("not an integer")
652 if value < 0 or value > 65535:
653 raise ValueError("not a uint16")
654 return value
656 @classmethod
657 def _as_uint32(cls, value):
658 if not isinstance(value, int):
659 raise ValueError("not an integer")
660 if value < 0 or value > 4294967295:
661 raise ValueError("not a uint32")
662 return value
664 @classmethod
665 def _as_uint48(cls, value):
666 if not isinstance(value, int):
667 raise ValueError("not an integer")
668 if value < 0 or value > 281474976710655:
669 raise ValueError("not a uint48")
670 return value
672 @classmethod
673 def _as_int(cls, value, low=None, high=None):
674 if not isinstance(value, int):
675 raise ValueError("not an integer")
676 if low is not None and value < low:
677 raise ValueError("value too small")
678 if high is not None and value > high:
679 raise ValueError("value too large")
680 return value
682 @classmethod
683 def _as_ipv4_address(cls, value):
684 if isinstance(value, str):
685 return dns.ipv4.canonicalize(value)
686 elif isinstance(value, bytes):
687 return dns.ipv4.inet_ntoa(value)
688 elif isinstance(value, ipaddress.IPv4Address):
689 return dns.ipv4.inet_ntoa(value.packed)
690 else:
691 raise ValueError("not an IPv4 address")
693 @classmethod
694 def _as_ipv6_address(cls, value):
695 if isinstance(value, str):
696 return dns.ipv6.canonicalize(value)
697 elif isinstance(value, bytes):
698 return dns.ipv6.inet_ntoa(value)
699 elif isinstance(value, ipaddress.IPv6Address):
700 return dns.ipv6.inet_ntoa(value.packed)
701 else:
702 raise ValueError("not an IPv6 address")
704 @classmethod
705 def _as_bool(cls, value):
706 if isinstance(value, bool):
707 return value
708 else:
709 raise ValueError("not a boolean")
711 @classmethod
712 def _as_ttl(cls, value):
713 if isinstance(value, int):
714 return cls._as_int(value, 0, dns.ttl.MAX_TTL)
715 elif isinstance(value, str):
716 return dns.ttl.from_text(value)
717 else:
718 raise ValueError("not a TTL")
720 @classmethod
721 def _as_tuple(cls, value, as_value):
722 try:
723 # For user convenience, if value is a singleton of the list
724 # element type, wrap it in a tuple.
725 return (as_value(value),)
726 except Exception:
727 # Otherwise, check each element of the iterable *value*
728 # against *as_value*.
729 return tuple(as_value(v) for v in value)
731 # Processing order
733 @classmethod
734 def _processing_order(cls, iterable):
735 items = list(iterable)
736 random.shuffle(items)
737 return items
740@dns.immutable.immutable
741class GenericRdata(Rdata):
742 """Generic Rdata Class
744 This class is used for rdata types for which we have no better
745 implementation. It implements the DNS "unknown RRs" scheme.
746 """
748 __slots__ = ["data"]
750 def __init__(
751 self,
752 rdclass: dns.rdataclass.RdataClass,
753 rdtype: dns.rdatatype.RdataType,
754 data: bytes,
755 ) -> None:
756 super().__init__(rdclass, rdtype)
757 self.data = data
759 def to_styled_text(
760 self,
761 style: RdataStyle,
762 ) -> str:
763 return rf"\# {len(self.data)} " + _styled_hexify(self.data, style)
765 @classmethod
766 def from_text(
767 cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None
768 ):
769 token = tok.get()
770 if not token.is_identifier() or token.value != r"\#":
771 raise dns.exception.SyntaxError(r"generic rdata does not start with \#")
772 length = tok.get_int()
773 hex = tok.concatenate_remaining_identifiers(True).encode()
774 data = binascii.unhexlify(hex)
775 if len(data) != length:
776 raise dns.exception.SyntaxError("generic rdata hex data has wrong length")
777 return cls(rdclass, rdtype, data)
779 def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
780 file.write(self.data)
782 def to_generic(self, origin: dns.name.Name | None = None) -> "GenericRdata":
783 return self
785 @classmethod
786 def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
787 return cls(rdclass, rdtype, parser.get_remaining())
790_rdata_classes: dict[tuple[dns.rdataclass.RdataClass, dns.rdatatype.RdataType], Any] = (
791 {}
792)
793_module_prefix = "dns.rdtypes"
794_dynamic_load_allowed = True
797def get_rdata_class(rdclass, rdtype, use_generic=True):
798 cls = _rdata_classes.get((rdclass, rdtype))
799 if not cls:
800 cls = _rdata_classes.get((dns.rdataclass.ANY, rdtype))
801 if not cls and _dynamic_load_allowed:
802 rdclass_text = dns.rdataclass.to_text(rdclass)
803 rdtype_text = dns.rdatatype.to_text(rdtype)
804 rdtype_text = rdtype_text.replace("-", "_")
805 try:
806 mod = import_module(
807 ".".join([_module_prefix, rdclass_text, rdtype_text])
808 )
809 cls = getattr(mod, rdtype_text)
810 _rdata_classes[(rdclass, rdtype)] = cls
811 except ImportError:
812 try:
813 mod = import_module(".".join([_module_prefix, "ANY", rdtype_text]))
814 cls = getattr(mod, rdtype_text)
815 _rdata_classes[(dns.rdataclass.ANY, rdtype)] = cls
816 _rdata_classes[(rdclass, rdtype)] = cls
817 except ImportError:
818 pass
819 if not cls and use_generic:
820 cls = GenericRdata
821 _rdata_classes[(rdclass, rdtype)] = cls
822 return cls
825def load_all_types(disable_dynamic_load=True):
826 """Load all rdata types for which dnspython has a non-generic implementation.
828 Normally dnspython loads DNS rdatatype implementations on demand, but in some
829 specialized cases loading all types at an application-controlled time is preferred.
831 :param disable_dynamic_load: If ``True`` (the default), dnspython will not
832 attempt to use its dynamic loading mechanism if an unknown type is
833 subsequently encountered, and will simply use the ``GenericRdata``
834 class.
835 :type disable_dynamic_load: bool
836 """
837 # Load class IN and ANY types.
838 for rdtype in dns.rdatatype.RdataType:
839 get_rdata_class(dns.rdataclass.IN, rdtype, False)
840 # Load the one non-ANY implementation we have in CH. Everything
841 # else in CH is an ANY type, and we'll discover those on demand but won't
842 # have to import anything.
843 get_rdata_class(dns.rdataclass.CH, dns.rdatatype.A, False)
844 if disable_dynamic_load:
845 # Now disable dynamic loading so any subsequent unknown type immediately becomes
846 # GenericRdata without a load attempt.
847 global _dynamic_load_allowed
848 _dynamic_load_allowed = False
851def from_text(
852 rdclass: dns.rdataclass.RdataClass | str,
853 rdtype: dns.rdatatype.RdataType | str,
854 tok: dns.tokenizer.Tokenizer | str,
855 origin: dns.name.Name | None = None,
856 relativize: bool = True,
857 relativize_to: dns.name.Name | None = None,
858 idna_codec: dns.name.IDNACodec | None = None,
859) -> Rdata:
860 """Build an rdata object from text format.
862 This function attempts to dynamically load a class which
863 implements the specified rdata class and type. If there is no
864 class-and-type-specific implementation, the GenericRdata class
865 is used.
867 Once a class is chosen, its from_text() class method is called
868 with the parameters to this function.
870 If *tok* is a ``str``, then a tokenizer is created and the string
871 is used as its input.
873 :param rdclass: The rdata class.
874 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
875 :param rdtype: The rdata type.
876 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
877 :param tok: The tokenizer or input string.
878 :type tok: :py:class:`dns.tokenizer.Tokenizer` or ``str``
879 :param origin: The origin to use for relative names.
880 :type origin: :py:class:`dns.name.Name` or ``None``
881 :param relativize: If ``True``, names will be relativized. Default is ``True``.
882 :type relativize: bool
883 :param relativize_to: The origin to use when relativizing. Defaults to *origin*.
884 :type relativize_to: :py:class:`dns.name.Name` or ``None``
885 :param idna_codec: The IDNA encoder/decoder to use when creating a tokenizer.
886 If ``None``, the default IDNA 2003 codec is used.
887 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
888 :returns: An instance of the chosen :py:class:`Rdata` subclass.
889 :rtype: :py:class:`Rdata`
891 """
892 if isinstance(tok, str):
893 tok = dns.tokenizer.Tokenizer(tok, idna_codec=idna_codec)
894 if not isinstance(tok, dns.tokenizer.Tokenizer):
895 raise ValueError("tok must be a string or a Tokenizer")
896 rdclass = dns.rdataclass.RdataClass.make(rdclass)
897 rdtype = dns.rdatatype.RdataType.make(rdtype)
898 cls = get_rdata_class(rdclass, rdtype)
899 assert cls is not None # for type checkers
900 with dns.exception.ExceptionWrapper(dns.exception.SyntaxError):
901 rdata = None
902 if cls != GenericRdata:
903 # peek at first token
904 token = tok.get()
905 tok.unget(token)
906 if token.is_identifier() and token.value == r"\#":
907 #
908 # Known type using the generic syntax. Extract the
909 # wire form from the generic syntax, and then run
910 # from_wire on it.
911 #
912 grdata = GenericRdata.from_text(
913 rdclass, rdtype, tok, origin, relativize, relativize_to
914 )
915 rdata = from_wire(rdclass, rdtype, grdata.data, 0, len(grdata.data))
916 #
917 # If this comparison isn't equal, then there must have been
918 # compressed names in the wire format, which is an error,
919 # there being no reasonable context to decompress with.
920 #
921 rwire = rdata.to_wire()
922 if rwire != grdata.data:
923 raise dns.exception.SyntaxError(
924 "compressed data in "
925 "generic syntax form "
926 "of known rdatatype"
927 )
928 # Names in wire format are absolute; relativize them as the
929 # text form would be.
930 relativize_origin = relativize_to or origin
931 if relativize and relativize_origin is not None:
932 rdata = from_wire(
933 rdclass,
934 rdtype,
935 grdata.data,
936 0,
937 len(grdata.data),
938 relativize_origin,
939 )
940 if rdata is None:
941 rdata = cls.from_text(
942 rdclass, rdtype, tok, origin, relativize, relativize_to
943 )
944 token = tok.get_eol_as_token()
945 if token.comment is not None:
946 object.__setattr__(rdata, "rdcomment", token.comment)
947 return rdata
950def from_wire_parser(
951 rdclass: dns.rdataclass.RdataClass | str,
952 rdtype: dns.rdatatype.RdataType | str,
953 parser: dns.wire.Parser,
954 origin: dns.name.Name | None = None,
955) -> Rdata:
956 """Build an rdata object from wire format
958 This function attempts to dynamically load a class which
959 implements the specified rdata class and type. If there is no
960 class-and-type-specific implementation, the GenericRdata class
961 is used.
963 Once a class is chosen, its from_wire() class method is called
964 with the parameters to this function.
966 :param rdclass: The rdata class.
967 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
968 :param rdtype: The rdata type.
969 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
970 :param parser: The wire format parser, restricted to the rdata length.
971 :type parser: :py:class:`dns.wire.Parser`
972 :param origin: If not ``None``, names will be relativized to this origin.
973 :type origin: :py:class:`dns.name.Name` or ``None``
974 :returns: An instance of the chosen :py:class:`Rdata` subclass.
975 :rtype: :py:class:`Rdata`
976 """
978 rdclass = dns.rdataclass.RdataClass.make(rdclass)
979 rdtype = dns.rdatatype.RdataType.make(rdtype)
980 cls = get_rdata_class(rdclass, rdtype)
981 assert cls is not None # for type checkers
982 with dns.exception.ExceptionWrapper(dns.exception.FormError):
983 return cls.from_wire_parser(rdclass, rdtype, parser, origin)
986def from_wire(
987 rdclass: dns.rdataclass.RdataClass | str,
988 rdtype: dns.rdatatype.RdataType | str,
989 wire: bytes,
990 current: int,
991 rdlen: int,
992 origin: dns.name.Name | None = None,
993) -> Rdata:
994 """Build an rdata object from wire format
996 This function attempts to dynamically load a class which
997 implements the specified rdata class and type. If there is no
998 class-and-type-specific implementation, the GenericRdata class
999 is used.
1001 Once a class is chosen, its from_wire() class method is called
1002 with the parameters to this function.
1004 :param rdclass: The rdata class.
1005 :type rdclass: :py:class:`dns.rdataclass.RdataClass` or ``str``
1006 :param rdtype: The rdata type.
1007 :type rdtype: :py:class:`dns.rdatatype.RdataType` or ``str``
1008 :param wire: The wire-format message.
1009 :type wire: bytes
1010 :param current: The offset of the start of the rdata in *wire*.
1011 :type current: int
1012 :param rdlen: The length of the rdata in wire format.
1013 :type rdlen: int
1014 :param origin: If not ``None``, names will be relativized to this origin.
1015 :type origin: :py:class:`dns.name.Name` or ``None``
1016 :returns: An instance of the chosen :py:class:`Rdata` subclass.
1017 :rtype: :py:class:`Rdata`
1018 """
1019 parser = dns.wire.Parser(wire, current)
1020 with parser.restrict_to(rdlen):
1021 return from_wire_parser(rdclass, rdtype, parser, origin)
1024class RdatatypeExists(dns.exception.DNSException):
1025 """DNS rdatatype already exists."""
1027 supp_kwargs = {"rdclass", "rdtype"}
1028 fmt = (
1029 "The rdata type with class {rdclass:d} and rdtype {rdtype:d} "
1030 + "already exists."
1031 )
1034def register_type(
1035 implementation: Any,
1036 rdtype: int,
1037 rdtype_text: str,
1038 is_singleton: bool = False,
1039 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
1040) -> None:
1041 """Dynamically register a module to handle an rdatatype.
1043 :param implementation: A subclass of :py:class:`dns.rdata.Rdata` implementing
1044 the type, or a module containing such a class named by its text form.
1045 :param rdtype: The rdatatype to register.
1046 :type rdtype: int
1047 :param rdtype_text: The textual form of the rdatatype.
1048 :type rdtype_text: str
1049 :param is_singleton: If ``True``, RRsets of this type can have only one member.
1050 :type is_singleton: bool
1051 :param rdclass: The rdata class, or ``dns.rdataclass.ANY`` for all classes.
1052 :type rdclass: :py:class:`dns.rdataclass.RdataClass`
1053 """
1055 rdtype = dns.rdatatype.RdataType.make(rdtype)
1056 existing_cls = get_rdata_class(rdclass, rdtype)
1057 if existing_cls != GenericRdata or dns.rdatatype.is_metatype(rdtype):
1058 raise RdatatypeExists(rdclass=rdclass, rdtype=rdtype)
1059 if isinstance(implementation, type) and issubclass(implementation, Rdata):
1060 impclass = implementation
1061 else:
1062 impclass = getattr(implementation, rdtype_text.replace("-", "_"))
1063 _rdata_classes[(rdclass, rdtype)] = impclass
1064 dns.rdatatype.register_type(rdtype, rdtype_text, is_singleton)