#$@%#$%!

Ngiler SH3LL 360
: / opt / hc_python / lib / python3.12 / site-packages / dns /

Information Server
MySQL : OFF Perl : OFF CURL : ON WGET : OFF PKEXEC : OFF

Directive	Local Value
IP Address	63.250.38.10
System	Linux premium90.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
User	likhxapw
PHP Version	8.3.30
Software	LiteSpeed
Doc root	Writable

Edit File : rdata.py \| Size : 30.29 KB
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license # Copyright (C) 2001-2017 Nominum, Inc. # # Permission to use, copy, modify, and distribute this software and its # documentation for any purpose with or without fee is hereby granted, # provided that the above copyright notice and this permission notice # appear in all copies. # # THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT # OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. """DNS rdata.""" import base64 import binascii import inspect import io import itertools import random from importlib import import_module from typing import Any, Dict, Optional, Tuple, Union import dns.exception import dns.immutable import dns.ipv4 import dns.ipv6 import dns.name import dns.rdataclass import dns.rdatatype import dns.tokenizer import dns.ttl import dns.wire _chunksize = 32 # We currently allow comparisons for rdata with relative names for backwards # compatibility, but in the future we will not, as these kinds of comparisons # can lead to subtle bugs if code is not carefully written. # # This switch allows the future behavior to be turned on so code can be # tested with it. _allow_relative_comparisons = True class NoRelativeRdataOrdering(dns.exception.DNSException): """An attempt was made to do an ordered comparison of one or more rdata with relative names. The only reliable way of sorting rdata is to use non-relativized rdata. """ def _wordbreak(data, chunksize=_chunksize, separator=b" "): """Break a binary string into chunks of chunksize characters separated by a space. """ if not chunksize: return data.decode() return separator.join( [data[i : i + chunksize] for i in range(0, len(data), chunksize)] ).decode() # pylint: disable=unused-argument def _hexify(data, chunksize=_chunksize, separator=b" ", kw): """Convert a binary string into its hex encoding, broken up into chunks of chunksize characters separated by a separator. """ return _wordbreak(binascii.hexlify(data), chunksize, separator) def _base64ify(data, chunksize=_chunksize, separator=b" ", kw): """Convert a binary string into its base64 encoding, broken up into chunks of chunksize characters separated by a separator. """ return _wordbreak(base64.b64encode(data), chunksize, separator) # pylint: enable=unused-argument __escaped = b'"\\' def _escapify(qstring): """Escape the characters in a quoted string which need it.""" if isinstance(qstring, str): qstring = qstring.encode() if not isinstance(qstring, bytearray): qstring = bytearray(qstring) text = "" for c in qstring: if c in __escaped: text += "\\" + chr(c) elif c >= 0x20 and c < 0x7F: text += chr(c) else: text += "\\%03d" % c return text def _truncate_bitmap(what): """Determine the index of greatest byte that isn't all zeros, and return the bitmap that contains all the bytes less than that index. """ for i in range(len(what) - 1, -1, -1): if what[i] != 0: return what[0 : i + 1] return what[0:1] # So we don't have to edit all the rdata classes... _constify = dns.immutable.constify @dns.immutable.immutable class Rdata: """Base class for all DNS rdata types.""" __slots__ = ["rdclass", "rdtype", "rdcomment"] def __init__(self, rdclass, rdtype): """Initialize an rdata. rdclass, an ``int`` is the rdataclass of the Rdata. rdtype, an ``int`` is the rdatatype of the Rdata. """ self.rdclass = self._as_rdataclass(rdclass) self.rdtype = self._as_rdatatype(rdtype) self.rdcomment = None def _get_all_slots(self): return itertools.chain.from_iterable( getattr(cls, "__slots__", []) for cls in self.__class__.__mro__ ) def __getstate__(self): # We used to try to do a tuple of all slots here, but it # doesn't work as self._all_slots isn't available at # __setstate__() time. Before that we tried to store a tuple # of __slots__, but that didn't work as it didn't store the # slots defined by ancestors. This older way didn't fail # outright, but ended up with partially broken objects, e.g. # if you unpickled an A RR it wouldn't have rdclass and rdtype # attributes, and would compare badly. state = {} for slot in self._get_all_slots(): state[slot] = getattr(self, slot) return state def __setstate__(self, state): for slot, val in state.items(): object.__setattr__(self, slot, val) if not hasattr(self, "rdcomment"): # Pickled rdata from 2.0.x might not have a rdcomment, so add # it if needed. object.__setattr__(self, "rdcomment", None) def covers(self) -> dns.rdatatype.RdataType: """Return the type a Rdata covers. DNS SIG/RRSIG rdatas apply to a specific type; this type is returned by the covers() function. If the rdata type is not SIG or RRSIG, dns.rdatatype.NONE is returned. This is useful when creating rdatasets, allowing the rdataset to contain only RRSIGs of a particular type, e.g. RRSIG(NS). Returns a ``dns.rdatatype.RdataType``. """ return dns.rdatatype.NONE def extended_rdatatype(self) -> int: """Return a 32-bit type value, the least significant 16 bits of which are the ordinary DNS type, and the upper 16 bits of which are the "covered" type, if any. Returns an ``int``. """ return self.covers() << 16 \| self.rdtype def to_text( self, origin: Optional[dns.name.Name] = None, relativize: bool = True, kw: Dict[str, Any], ) -> str: """Convert an rdata to text format. Returns a ``str``. """ raise NotImplementedError # pragma: no cover def _to_wire( self, file: Optional[Any], compress: Optional[dns.name.CompressType] = None, origin: Optional[dns.name.Name] = None, canonicalize: bool = False, ) -> None: raise NotImplementedError # pragma: no cover def to_wire( self, file: Optional[Any] = None, compress: Optional[dns.name.CompressType] = None, origin: Optional[dns.name.Name] = None, canonicalize: bool = False, ) -> Optional[bytes]: """Convert an rdata to wire format. Returns a ``bytes`` if no output file was specified, or ``None`` otherwise. """ if file: # We call _to_wire() and then return None explicitly instead of # of just returning the None from _to_wire() as mypy's func-returns-value # unhelpfully errors out with "error: "_to_wire" of "Rdata" does not return # a value (it only ever returns None)" self._to_wire(file, compress, origin, canonicalize) return None else: f = io.BytesIO() self._to_wire(f, compress, origin, canonicalize) return f.getvalue() def to_generic( self, origin: Optional[dns.name.Name] = None ) -> "dns.rdata.GenericRdata": """Creates a dns.rdata.GenericRdata equivalent of this rdata. Returns a ``dns.rdata.GenericRdata``. """ return dns.rdata.GenericRdata( self.rdclass, self.rdtype, self.to_wire(origin=origin) ) def to_digestable(self, origin: Optional[dns.name.Name] = None) -> bytes: """Convert rdata to a format suitable for digesting in hashes. This is also the DNSSEC canonical form. Returns a ``bytes``. """ wire = self.to_wire(origin=origin, canonicalize=True) assert wire is not None # for mypy return wire def __repr__(self): covers = self.covers() if covers == dns.rdatatype.NONE: ctext = "" else: ctext = "(" + dns.rdatatype.to_text(covers) + ")" return ( "<DNS " + dns.rdataclass.to_text(self.rdclass) + " " + dns.rdatatype.to_text(self.rdtype) + ctext + " rdata: " + str(self) + ">" ) def __str__(self): return self.to_text() def _cmp(self, other): """Compare an rdata with another rdata of the same rdtype and rdclass. For rdata with only absolute names: Return < 0 if self < other in the DNSSEC ordering, 0 if self == other, and > 0 if self > other. For rdata with at least one relative names: The rdata sorts before any rdata with only absolute names. When compared with another relative rdata, all names are made absolute as if they were relative to the root, as the proper origin is not available. While this creates a stable ordering, it is NOT guaranteed to be the DNSSEC ordering. In the future, all ordering comparisons for rdata with relative names will be disallowed. """ try: our = self.to_digestable() our_relative = False except dns.name.NeedAbsoluteNameOrOrigin: if _allow_relative_comparisons: our = self.to_digestable(dns.name.root) our_relative = True try: their = other.to_digestable() their_relative = False except dns.name.NeedAbsoluteNameOrOrigin: if _allow_relative_comparisons: their = other.to_digestable(dns.name.root) their_relative = True if _allow_relative_comparisons: if our_relative != their_relative: # For the purpose of comparison, all rdata with at least one # relative name is less than an rdata with only absolute names. if our_relative: return -1 else: return 1 elif our_relative or their_relative: raise NoRelativeRdataOrdering if our == their: return 0 elif our > their: return 1 else: return -1 def __eq__(self, other): if not isinstance(other, Rdata): return False if self.rdclass != other.rdclass or self.rdtype != other.rdtype: return False our_relative = False their_relative = False try: our = self.to_digestable() except dns.name.NeedAbsoluteNameOrOrigin: our = self.to_digestable(dns.name.root) our_relative = True try: their = other.to_digestable() except dns.name.NeedAbsoluteNameOrOrigin: their = other.to_digestable(dns.name.root) their_relative = True if our_relative != their_relative: return False return our == their def __ne__(self, other): if not isinstance(other, Rdata): return True if self.rdclass != other.rdclass or self.rdtype != other.rdtype: return True return not self.__eq__(other) def __lt__(self, other): if ( not isinstance(other, Rdata) or self.rdclass != other.rdclass or self.rdtype != other.rdtype ): return NotImplemented return self._cmp(other) < 0 def __le__(self, other): if ( not isinstance(other, Rdata) or self.rdclass != other.rdclass or self.rdtype != other.rdtype ): return NotImplemented return self._cmp(other) <= 0 def __ge__(self, other): if ( not isinstance(other, Rdata) or self.rdclass != other.rdclass or self.rdtype != other.rdtype ): return NotImplemented return self._cmp(other) >= 0 def __gt__(self, other): if ( not isinstance(other, Rdata) or self.rdclass != other.rdclass or self.rdtype != other.rdtype ): return NotImplemented return self._cmp(other) > 0 def __hash__(self): return hash(self.to_digestable(dns.name.root)) @classmethod def from_text( cls, rdclass: dns.rdataclass.RdataClass, rdtype: dns.rdatatype.RdataType, tok: dns.tokenizer.Tokenizer, origin: Optional[dns.name.Name] = None, relativize: bool = True, relativize_to: Optional[dns.name.Name] = None, ) -> "Rdata": raise NotImplementedError # pragma: no cover @classmethod def from_wire_parser( cls, rdclass: dns.rdataclass.RdataClass, rdtype: dns.rdatatype.RdataType, parser: dns.wire.Parser, origin: Optional[dns.name.Name] = None, ) -> "Rdata": raise NotImplementedError # pragma: no cover def replace(self, kwargs: Any) -> "Rdata": """ Create a new Rdata instance based on the instance replace was invoked on. It is possible to pass different parameters to override the corresponding properties of the base Rdata. Any field specific to the Rdata type can be replaced, but the rdtype and rdclass fields cannot. Returns an instance of the same Rdata subclass as self. """ # Get the constructor parameters. parameters = inspect.signature(self.__init__).parameters # type: ignore # Ensure that all of the arguments correspond to valid fields. # Don't allow rdclass or rdtype to be changed, though. for key in kwargs: if key == "rdcomment": continue if key not in parameters: raise AttributeError( f"'{self.__class__.__name__}' object has no attribute '{key}'" ) if key in ("rdclass", "rdtype"): raise AttributeError( f"Cannot overwrite '{self.__class__.__name__}' attribute '{key}'" ) # Construct the parameter list. For each field, use the value in # kwargs if present, and the current value otherwise. args = (kwargs.get(key, getattr(self, key)) for key in parameters) # Create, validate, and return the new object. rd = self.__class__(args) # The comment is not set in the constructor, so give it special # handling. rdcomment = kwargs.get("rdcomment", self.rdcomment) if rdcomment is not None: object.__setattr__(rd, "rdcomment", rdcomment) return rd # Type checking and conversion helpers. These are class methods as # they don't touch object state and may be useful to others. @classmethod def _as_rdataclass(cls, value): return dns.rdataclass.RdataClass.make(value) @classmethod def _as_rdatatype(cls, value): return dns.rdatatype.RdataType.make(value) @classmethod def _as_bytes( cls, value: Any, encode: bool = False, max_length: Optional[int] = None, empty_ok: bool = True, ) -> bytes: if encode and isinstance(value, str): bvalue = value.encode() elif isinstance(value, bytearray): bvalue = bytes(value) elif isinstance(value, bytes): bvalue = value else: raise ValueError("not bytes") if max_length is not None and len(bvalue) > max_length: raise ValueError("too long") if not empty_ok and len(bvalue) == 0: raise ValueError("empty bytes not allowed") return bvalue @classmethod def _as_name(cls, value): # Note that proper name conversion (e.g. with origin and IDNA # awareness) is expected to be done via from_text. This is just # a simple thing for people invoking the constructor directly. if isinstance(value, str): return dns.name.from_text(value) elif not isinstance(value, dns.name.Name): raise ValueError("not a name") return value @classmethod def _as_uint8(cls, value): if not isinstance(value, int): raise ValueError("not an integer") if value < 0 or value > 255: raise ValueError("not a uint8") return value @classmethod def _as_uint16(cls, value): if not isinstance(value, int): raise ValueError("not an integer") if value < 0 or value > 65535: raise ValueError("not a uint16") return value @classmethod def _as_uint32(cls, value): if not isinstance(value, int): raise ValueError("not an integer") if value < 0 or value > 4294967295: raise ValueError("not a uint32") return value @classmethod def _as_uint48(cls, value): if not isinstance(value, int): raise ValueError("not an integer") if value < 0 or value > 281474976710655: raise ValueError("not a uint48") return value @classmethod def _as_int(cls, value, low=None, high=None): if not isinstance(value, int): raise ValueError("not an integer") if low is not None and value < low: raise ValueError("value too small") if high is not None and value > high: raise ValueError("value too large") return value @classmethod def _as_ipv4_address(cls, value): if isinstance(value, str): return dns.ipv4.canonicalize(value) elif isinstance(value, bytes): return dns.ipv4.inet_ntoa(value) else: raise ValueError("not an IPv4 address") @classmethod def _as_ipv6_address(cls, value): if isinstance(value, str): return dns.ipv6.canonicalize(value) elif isinstance(value, bytes): return dns.ipv6.inet_ntoa(value) else: raise ValueError("not an IPv6 address") @classmethod def _as_bool(cls, value): if isinstance(value, bool): return value else: raise ValueError("not a boolean") @classmethod def _as_ttl(cls, value): if isinstance(value, int): return cls._as_int(value, 0, dns.ttl.MAX_TTL) elif isinstance(value, str): return dns.ttl.from_text(value) else: raise ValueError("not a TTL") @classmethod def _as_tuple(cls, value, as_value): try: # For user convenience, if value is a singleton of the list # element type, wrap it in a tuple. return (as_value(value),) except Exception: # Otherwise, check each element of the iterable value* # against as_value. return tuple(as_value(v) for v in value) # Processing order @classmethod def _processing_order(cls, iterable): items = list(iterable) random.shuffle(items) return items @dns.immutable.immutable class GenericRdata(Rdata): """Generic Rdata Class This class is used for rdata types for which we have no better implementation. It implements the DNS "unknown RRs" scheme. """ __slots__ = ["data"] def __init__(self, rdclass, rdtype, data): super().__init__(rdclass, rdtype) self.data = data def to_text( self, origin: Optional[dns.name.Name] = None, relativize: bool = True, kw: Dict[str, Any], ) -> str: return r"\# %d " % len(self.data) + _hexify(self.data, kw) @classmethod def from_text( cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None ): token = tok.get() if not token.is_identifier() or token.value != r"\#": raise dns.exception.SyntaxError(r"generic rdata does not start with \#") length = tok.get_int() hex = tok.concatenate_remaining_identifiers(True).encode() data = binascii.unhexlify(hex) if len(data) != length: raise dns.exception.SyntaxError("generic rdata hex data has wrong length") return cls(rdclass, rdtype, data) def _to_wire(self, file, compress=None, origin=None, canonicalize=False): file.write(self.data) @classmethod def from_wire_parser(cls, rdclass, rdtype, parser, origin=None): return cls(rdclass, rdtype, parser.get_remaining()) _rdata_classes: Dict[Tuple[dns.rdataclass.RdataClass, dns.rdatatype.RdataType], Any] = ( {} ) _module_prefix = "dns.rdtypes" _dynamic_load_allowed = True def get_rdata_class(rdclass, rdtype, use_generic=True): cls = _rdata_classes.get((rdclass, rdtype)) if not cls: cls = _rdata_classes.get((dns.rdatatype.ANY, rdtype)) if not cls and _dynamic_load_allowed: rdclass_text = dns.rdataclass.to_text(rdclass) rdtype_text = dns.rdatatype.to_text(rdtype) rdtype_text = rdtype_text.replace("-", "_") try: mod = import_module( ".".join([_module_prefix, rdclass_text, rdtype_text]) ) cls = getattr(mod, rdtype_text) _rdata_classes[(rdclass, rdtype)] = cls except ImportError: try: mod = import_module(".".join([_module_prefix, "ANY", rdtype_text])) cls = getattr(mod, rdtype_text) _rdata_classes[(dns.rdataclass.ANY, rdtype)] = cls _rdata_classes[(rdclass, rdtype)] = cls except ImportError: pass if not cls and use_generic: cls = GenericRdata _rdata_classes[(rdclass, rdtype)] = cls return cls def load_all_types(disable_dynamic_load=True): """Load all rdata types for which dnspython has a non-generic implementation. Normally dnspython loads DNS rdatatype implementations on demand, but in some specialized cases loading all types at an application-controlled time is preferred. If disable_dynamic_load, a ``bool``, is ``True`` then dnspython will not attempt to use its dynamic loading mechanism if an unknown type is subsequently encountered, and will simply use the ``GenericRdata`` class. """ # Load class IN and ANY types. for rdtype in dns.rdatatype.RdataType: get_rdata_class(dns.rdataclass.IN, rdtype, False) # Load the one non-ANY implementation we have in CH. Everything # else in CH is an ANY type, and we'll discover those on demand but won't # have to import anything. get_rdata_class(dns.rdataclass.CH, dns.rdatatype.A, False) if disable_dynamic_load: # Now disable dynamic loading so any subsequent unknown type immediately becomes # GenericRdata without a load attempt. global _dynamic_load_allowed _dynamic_load_allowed = False def from_text( rdclass: Union[dns.rdataclass.RdataClass, str], rdtype: Union[dns.rdatatype.RdataType, str], tok: Union[dns.tokenizer.Tokenizer, str], origin: Optional[dns.name.Name] = None, relativize: bool = True, relativize_to: Optional[dns.name.Name] = None, idna_codec: Optional[dns.name.IDNACodec] = None, ) -> Rdata: """Build an rdata object from text format. This function attempts to dynamically load a class which implements the specified rdata class and type. If there is no class-and-type-specific implementation, the GenericRdata class is used. Once a class is chosen, its from_text() class method is called with the parameters to this function. If tok is a ``str``, then a tokenizer is created and the string is used as its input. rdclass, a ``dns.rdataclass.RdataClass`` or ``str``, the rdataclass. rdtype, a ``dns.rdatatype.RdataType`` or ``str``, the rdatatype. tok, a ``dns.tokenizer.Tokenizer`` or a ``str``. origin, a ``dns.name.Name`` (or ``None``), the origin to use for relative names. relativize, a ``bool``. If true, name will be relativized. relativize_to, a ``dns.name.Name`` (or ``None``), the origin to use when relativizing names. If not set, the origin value will be used. idna_codec, a ``dns.name.IDNACodec``, specifies the IDNA encoder/decoder to use if a tokenizer needs to be created. If ``None``, the default IDNA 2003 encoder/decoder is used. If a tokenizer is not created, then the codec associated with the tokenizer is the one that is used. Returns an instance of the chosen Rdata subclass. """ if isinstance(tok, str): tok = dns.tokenizer.Tokenizer(tok, idna_codec=idna_codec) rdclass = dns.rdataclass.RdataClass.make(rdclass) rdtype = dns.rdatatype.RdataType.make(rdtype) cls = get_rdata_class(rdclass, rdtype) with dns.exception.ExceptionWrapper(dns.exception.SyntaxError): rdata = None if cls != GenericRdata: # peek at first token token = tok.get() tok.unget(token) if token.is_identifier() and token.value == r"\#": # # Known type using the generic syntax. Extract the # wire form from the generic syntax, and then run # from_wire on it. # grdata = GenericRdata.from_text( rdclass, rdtype, tok, origin, relativize, relativize_to ) rdata = from_wire( rdclass, rdtype, grdata.data, 0, len(grdata.data), origin ) # # If this comparison isn't equal, then there must have been # compressed names in the wire format, which is an error, # there being no reasonable context to decompress with. # rwire = rdata.to_wire() if rwire != grdata.data: raise dns.exception.SyntaxError( "compressed data in " "generic syntax form " "of known rdatatype" ) if rdata is None: rdata = cls.from_text( rdclass, rdtype, tok, origin, relativize, relativize_to ) token = tok.get_eol_as_token() if token.comment is not None: object.__setattr__(rdata, "rdcomment", token.comment) return rdata def from_wire_parser( rdclass: Union[dns.rdataclass.RdataClass, str], rdtype: Union[dns.rdatatype.RdataType, str], parser: dns.wire.Parser, origin: Optional[dns.name.Name] = None, ) -> Rdata: """Build an rdata object from wire format This function attempts to dynamically load a class which implements the specified rdata class and type. If there is no class-and-type-specific implementation, the GenericRdata class is used. Once a class is chosen, its from_wire() class method is called with the parameters to this function. rdclass, a ``dns.rdataclass.RdataClass`` or ``str``, the rdataclass. rdtype, a ``dns.rdatatype.RdataType`` or ``str``, the rdatatype. parser, a ``dns.wire.Parser``, the parser, which should be restricted to the rdata length. origin, a ``dns.name.Name`` (or ``None``). If not ``None``, then names will be relativized to this origin. Returns an instance of the chosen Rdata subclass. """ rdclass = dns.rdataclass.RdataClass.make(rdclass) rdtype = dns.rdatatype.RdataType.make(rdtype) cls = get_rdata_class(rdclass, rdtype) with dns.exception.ExceptionWrapper(dns.exception.FormError): return cls.from_wire_parser(rdclass, rdtype, parser, origin) def from_wire( rdclass: Union[dns.rdataclass.RdataClass, str], rdtype: Union[dns.rdatatype.RdataType, str], wire: bytes, current: int, rdlen: int, origin: Optional[dns.name.Name] = None, ) -> Rdata: """Build an rdata object from wire format This function attempts to dynamically load a class which implements the specified rdata class and type. If there is no class-and-type-specific implementation, the GenericRdata class is used. Once a class is chosen, its from_wire() class method is called with the parameters to this function. rdclass, an ``int``, the rdataclass. rdtype, an ``int``, the rdatatype. wire, a ``bytes``, the wire-format message. current, an ``int``, the offset in wire of the beginning of the rdata. rdlen, an ``int``, the length of the wire-format rdata origin, a ``dns.name.Name`` (or ``None``). If not ``None``, then names will be relativized to this origin. Returns an instance of the chosen Rdata subclass. """ parser = dns.wire.Parser(wire, current) with parser.restrict_to(rdlen): return from_wire_parser(rdclass, rdtype, parser, origin) class RdatatypeExists(dns.exception.DNSException): """DNS rdatatype already exists.""" supp_kwargs = {"rdclass", "rdtype"} fmt = ( "The rdata type with class {rdclass:d} and rdtype {rdtype:d} " + "already exists." ) def register_type( implementation: Any, rdtype: int, rdtype_text: str, is_singleton: bool = False, rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN, ) -> None: """Dynamically register a module to handle an rdatatype. implementation, a module implementing the type in the usual dnspython way. rdtype, an ``int``, the rdatatype to register. rdtype_text, a ``str``, the textual form of the rdatatype. is_singleton, a ``bool``, indicating if the type is a singleton (i.e. RRsets of the type can have only one member.) rdclass, the rdataclass of the type, or ``dns.rdataclass.ANY`` if it applies to all classes. """ rdtype = dns.rdatatype.RdataType.make(rdtype) existing_cls = get_rdata_class(rdclass, rdtype) if existing_cls != GenericRdata or dns.rdatatype.is_metatype(rdtype): raise RdatatypeExists(rdclass=rdclass, rdtype=rdtype) _rdata_classes[(rdclass, rdtype)] = getattr( implementation, rdtype_text.replace("-", "_") ) dns.rdatatype.register_type(rdtype, rdtype_text, is_singleton) Back

Edit File : rdata.py | Size : 30.29 KB

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

# Copyright (C) 2001-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

"""DNS rdata."""

import base64
import binascii
import inspect
import io
import itertools
import random
from importlib import import_module
from typing import Any, Dict, Optional, Tuple, Union

import dns.exception
import dns.immutable
import dns.ipv4
import dns.ipv6
import dns.name
import dns.rdataclass
import dns.rdatatype
import dns.tokenizer
import dns.ttl
import dns.wire

_chunksize = 32

# We currently allow comparisons for rdata with relative names for backwards
# compatibility, but in the future we will not, as these kinds of comparisons
# can lead to subtle bugs if code is not carefully written.
#
# This switch allows the future behavior to be turned on so code can be
# tested with it.
_allow_relative_comparisons = True

class NoRelativeRdataOrdering(dns.exception.DNSException):
    """An attempt was made to do an ordered comparison of one or more
    rdata with relative names.  The only reliable way of sorting rdata
    is to use non-relativized rdata.

"""

def _wordbreak(data, chunksize=_chunksize, separator=b" "):
    """Break a binary string into chunks of chunksize characters separated by
    a space.
    """

if not chunksize:
        return data.decode()
    return separator.join(
        [data[i : i + chunksize] for i in range(0, len(data), chunksize)]
    ).decode()

# pylint: disable=unused-argument

def _hexify(data, chunksize=_chunksize, separator=b" ", **kw):
    """Convert a binary string into its hex encoding, broken up into chunks
    of chunksize characters separated by a separator.
    """

return _wordbreak(binascii.hexlify(data), chunksize, separator)

def _base64ify(data, chunksize=_chunksize, separator=b" ", **kw):
    """Convert a binary string into its base64 encoding, broken up into chunks
    of chunksize characters separated by a separator.
    """

return _wordbreak(base64.b64encode(data), chunksize, separator)

# pylint: enable=unused-argument

__escaped = b'"\\'

def _escapify(qstring):
    """Escape the characters in a quoted string which need it."""

if isinstance(qstring, str):
        qstring = qstring.encode()
    if not isinstance(qstring, bytearray):
        qstring = bytearray(qstring)

text = ""
    for c in qstring:
        if c in __escaped:
            text += "\\" + chr(c)
        elif c >= 0x20 and c < 0x7F:
            text += chr(c)
        else:
            text += "\\%03d" % c
    return text

def _truncate_bitmap(what):
    """Determine the index of greatest byte that isn't all zeros, and
    return the bitmap that contains all the bytes less than that index.
    """

for i in range(len(what) - 1, -1, -1):
        if what[i] != 0:
            return what[0 : i + 1]
    return what[0:1]

# So we don't have to edit all the rdata classes...
_constify = dns.immutable.constify

@dns.immutable.immutable
class Rdata:
    """Base class for all DNS rdata types."""

__slots__ = ["rdclass", "rdtype", "rdcomment"]

def __init__(self, rdclass, rdtype):
        """Initialize an rdata.

*rdclass*, an ``int`` is the rdataclass of the Rdata.

*rdtype*, an ``int`` is the rdatatype of the Rdata.
        """

self.rdclass = self._as_rdataclass(rdclass)
        self.rdtype = self._as_rdatatype(rdtype)
        self.rdcomment = None

def _get_all_slots(self):
        return itertools.chain.from_iterable(
            getattr(cls, "__slots__", []) for cls in self.__class__.__mro__
        )

def __getstate__(self):
        # We used to try to do a tuple of all slots here, but it
        # doesn't work as self._all_slots isn't available at
        # __setstate__() time.  Before that we tried to store a tuple
        # of __slots__, but that didn't work as it didn't store the
        # slots defined by ancestors.  This older way didn't fail
        # outright, but ended up with partially broken objects, e.g.
        # if you unpickled an A RR it wouldn't have rdclass and rdtype
        # attributes, and would compare badly.
        state = {}
        for slot in self._get_all_slots():
            state[slot] = getattr(self, slot)
        return state

def __setstate__(self, state):
        for slot, val in state.items():
            object.__setattr__(self, slot, val)
        if not hasattr(self, "rdcomment"):
            # Pickled rdata from 2.0.x might not have a rdcomment, so add
            # it if needed.
            object.__setattr__(self, "rdcomment", None)

def covers(self) -> dns.rdatatype.RdataType:
        """Return the type a Rdata covers.

DNS SIG/RRSIG rdatas apply to a specific type; this type is
        returned by the covers() function.  If the rdata type is not
        SIG or RRSIG, dns.rdatatype.NONE is returned.  This is useful when
        creating rdatasets, allowing the rdataset to contain only RRSIGs
        of a particular type, e.g. RRSIG(NS).

Returns a ``dns.rdatatype.RdataType``.
        """

return dns.rdatatype.NONE

def extended_rdatatype(self) -> int:
        """Return a 32-bit type value, the least significant 16 bits of
        which are the ordinary DNS type, and the upper 16 bits of which are
        the "covered" type, if any.

Returns an ``int``.
        """

return self.covers() << 16 | self.rdtype

def to_text(
        self,
        origin: Optional[dns.name.Name] = None,
        relativize: bool = True,
        **kw: Dict[str, Any],
    ) -> str:
        """Convert an rdata to text format.

Returns a ``str``.
        """

raise NotImplementedError  # pragma: no cover

def _to_wire(
        self,
        file: Optional[Any],
        compress: Optional[dns.name.CompressType] = None,
        origin: Optional[dns.name.Name] = None,
        canonicalize: bool = False,
    ) -> None:
        raise NotImplementedError  # pragma: no cover

def to_wire(
        self,
        file: Optional[Any] = None,
        compress: Optional[dns.name.CompressType] = None,
        origin: Optional[dns.name.Name] = None,
        canonicalize: bool = False,
    ) -> Optional[bytes]:
        """Convert an rdata to wire format.

Returns a ``bytes`` if no output file was specified, or ``None`` otherwise.
        """

if file:
            # We call _to_wire() and then return None explicitly instead of
            # of just returning the None from _to_wire() as mypy's func-returns-value
            # unhelpfully errors out with "error: "_to_wire" of "Rdata" does not return
            # a value (it only ever returns None)"
            self._to_wire(file, compress, origin, canonicalize)
            return None
        else:
            f = io.BytesIO()
            self._to_wire(f, compress, origin, canonicalize)
            return f.getvalue()

def to_generic(
        self, origin: Optional[dns.name.Name] = None
    ) -> "dns.rdata.GenericRdata":
        """Creates a dns.rdata.GenericRdata equivalent of this rdata.

Returns a ``dns.rdata.GenericRdata``.
        """
        return dns.rdata.GenericRdata(
            self.rdclass, self.rdtype, self.to_wire(origin=origin)
        )

def to_digestable(self, origin: Optional[dns.name.Name] = None) -> bytes:
        """Convert rdata to a format suitable for digesting in hashes.  This
        is also the DNSSEC canonical form.

Returns a ``bytes``.
        """
        wire = self.to_wire(origin=origin, canonicalize=True)
        assert wire is not None  # for mypy
        return wire

def __repr__(self):
        covers = self.covers()
        if covers == dns.rdatatype.NONE:
            ctext = ""
        else:
            ctext = "(" + dns.rdatatype.to_text(covers) + ")"
        return (
            "<DNS "
            + dns.rdataclass.to_text(self.rdclass)
            + " "
            + dns.rdatatype.to_text(self.rdtype)
            + ctext
            + " rdata: "
            + str(self)
            + ">"
        )

def __str__(self):
        return self.to_text()

def _cmp(self, other):
        """Compare an rdata with another rdata of the same rdtype and
        rdclass.

For rdata with only absolute names:
            Return < 0 if self < other in the DNSSEC ordering, 0 if self
            == other, and > 0 if self > other.
        For rdata with at least one relative names:
            The rdata sorts before any rdata with only absolute names.
            When compared with another relative rdata, all names are
            made absolute as if they were relative to the root, as the
            proper origin is not available.  While this creates a stable
            ordering, it is NOT guaranteed to be the DNSSEC ordering.
            In the future, all ordering comparisons for rdata with
            relative names will be disallowed.
        """
        try:
            our = self.to_digestable()
            our_relative = False
        except dns.name.NeedAbsoluteNameOrOrigin:
            if _allow_relative_comparisons:
                our = self.to_digestable(dns.name.root)
            our_relative = True
        try:
            their = other.to_digestable()
            their_relative = False
        except dns.name.NeedAbsoluteNameOrOrigin:
            if _allow_relative_comparisons:
                their = other.to_digestable(dns.name.root)
            their_relative = True
        if _allow_relative_comparisons:
            if our_relative != their_relative:
                # For the purpose of comparison, all rdata with at least one
                # relative name is less than an rdata with only absolute names.
                if our_relative:
                    return -1
                else:
                    return 1
        elif our_relative or their_relative:
            raise NoRelativeRdataOrdering
        if our == their:
            return 0
        elif our > their:
            return 1
        else:
            return -1

def __eq__(self, other):
        if not isinstance(other, Rdata):
            return False
        if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return False
        our_relative = False
        their_relative = False
        try:
            our = self.to_digestable()
        except dns.name.NeedAbsoluteNameOrOrigin:
            our = self.to_digestable(dns.name.root)
            our_relative = True
        try:
            their = other.to_digestable()
        except dns.name.NeedAbsoluteNameOrOrigin:
            their = other.to_digestable(dns.name.root)
            their_relative = True
        if our_relative != their_relative:
            return False
        return our == their

def __ne__(self, other):
        if not isinstance(other, Rdata):
            return True
        if self.rdclass != other.rdclass or self.rdtype != other.rdtype:
            return True
        return not self.__eq__(other)

def __lt__(self, other):
        if (
            not isinstance(other, Rdata)
            or self.rdclass != other.rdclass
            or self.rdtype != other.rdtype
        ):
            return NotImplemented
        return self._cmp(other) < 0

def __le__(self, other):
        if (
            not isinstance(other, Rdata)
            or self.rdclass != other.rdclass
            or self.rdtype != other.rdtype
        ):
            return NotImplemented
        return self._cmp(other) <= 0

def __ge__(self, other):
        if (
            not isinstance(other, Rdata)
            or self.rdclass != other.rdclass
            or self.rdtype != other.rdtype
        ):
            return NotImplemented
        return self._cmp(other) >= 0

def __gt__(self, other):
        if (
            not isinstance(other, Rdata)
            or self.rdclass != other.rdclass
            or self.rdtype != other.rdtype
        ):
            return NotImplemented
        return self._cmp(other) > 0

def __hash__(self):
        return hash(self.to_digestable(dns.name.root))

@classmethod
    def from_text(
        cls,
        rdclass: dns.rdataclass.RdataClass,
        rdtype: dns.rdatatype.RdataType,
        tok: dns.tokenizer.Tokenizer,
        origin: Optional[dns.name.Name] = None,
        relativize: bool = True,
        relativize_to: Optional[dns.name.Name] = None,
    ) -> "Rdata":
        raise NotImplementedError  # pragma: no cover

@classmethod
    def from_wire_parser(
        cls,
        rdclass: dns.rdataclass.RdataClass,
        rdtype: dns.rdatatype.RdataType,
        parser: dns.wire.Parser,
        origin: Optional[dns.name.Name] = None,
    ) -> "Rdata":
        raise NotImplementedError  # pragma: no cover

def replace(self, **kwargs: Any) -> "Rdata":
        """
        Create a new Rdata instance based on the instance replace was
        invoked on. It is possible to pass different parameters to
        override the corresponding properties of the base Rdata.

Any field specific to the Rdata type can be replaced, but the
        *rdtype* and *rdclass* fields cannot.

Returns an instance of the same Rdata subclass as *self*.
        """

# Get the constructor parameters.
        parameters = inspect.signature(self.__init__).parameters  # type: ignore

# Ensure that all of the arguments correspond to valid fields.
        # Don't allow rdclass or rdtype to be changed, though.
        for key in kwargs:
            if key == "rdcomment":
                continue
            if key not in parameters:
                raise AttributeError(
                    f"'{self.__class__.__name__}' object has no attribute '{key}'"
                )
            if key in ("rdclass", "rdtype"):
                raise AttributeError(
                    f"Cannot overwrite '{self.__class__.__name__}' attribute '{key}'"
                )

# Construct the parameter list.  For each field, use the value in
        # kwargs if present, and the current value otherwise.
        args = (kwargs.get(key, getattr(self, key)) for key in parameters)

# Create, validate, and return the new object.
        rd = self.__class__(*args)
        # The comment is not set in the constructor, so give it special
        # handling.
        rdcomment = kwargs.get("rdcomment", self.rdcomment)
        if rdcomment is not None:
            object.__setattr__(rd, "rdcomment", rdcomment)
        return rd

# Type checking and conversion helpers.  These are class methods as
    # they don't touch object state and may be useful to others.

@classmethod
    def _as_rdataclass(cls, value):
        return dns.rdataclass.RdataClass.make(value)

@classmethod
    def _as_rdatatype(cls, value):
        return dns.rdatatype.RdataType.make(value)

@classmethod
    def _as_bytes(
        cls,
        value: Any,
        encode: bool = False,
        max_length: Optional[int] = None,
        empty_ok: bool = True,
    ) -> bytes:
        if encode and isinstance(value, str):
            bvalue = value.encode()
        elif isinstance(value, bytearray):
            bvalue = bytes(value)
        elif isinstance(value, bytes):
            bvalue = value
        else:
            raise ValueError("not bytes")
        if max_length is not None and len(bvalue) > max_length:
            raise ValueError("too long")
        if not empty_ok and len(bvalue) == 0:
            raise ValueError("empty bytes not allowed")
        return bvalue

@classmethod
    def _as_name(cls, value):
        # Note that proper name conversion (e.g. with origin and IDNA
        # awareness) is expected to be done via from_text.  This is just
        # a simple thing for people invoking the constructor directly.
        if isinstance(value, str):
            return dns.name.from_text(value)
        elif not isinstance(value, dns.name.Name):
            raise ValueError("not a name")
        return value

@classmethod
    def _as_uint8(cls, value):
        if not isinstance(value, int):
            raise ValueError("not an integer")
        if value < 0 or value > 255:
            raise ValueError("not a uint8")
        return value

@classmethod
    def _as_uint16(cls, value):
        if not isinstance(value, int):
            raise ValueError("not an integer")
        if value < 0 or value > 65535:
            raise ValueError("not a uint16")
        return value

@classmethod
    def _as_uint32(cls, value):
        if not isinstance(value, int):
            raise ValueError("not an integer")
        if value < 0 or value > 4294967295:
            raise ValueError("not a uint32")
        return value

@classmethod
    def _as_uint48(cls, value):
        if not isinstance(value, int):
            raise ValueError("not an integer")
        if value < 0 or value > 281474976710655:
            raise ValueError("not a uint48")
        return value

@classmethod
    def _as_int(cls, value, low=None, high=None):
        if not isinstance(value, int):
            raise ValueError("not an integer")
        if low is not None and value < low:
            raise ValueError("value too small")
        if high is not None and value > high:
            raise ValueError("value too large")
        return value

@classmethod
    def _as_ipv4_address(cls, value):
        if isinstance(value, str):
            return dns.ipv4.canonicalize(value)
        elif isinstance(value, bytes):
            return dns.ipv4.inet_ntoa(value)
        else:
            raise ValueError("not an IPv4 address")

@classmethod
    def _as_ipv6_address(cls, value):
        if isinstance(value, str):
            return dns.ipv6.canonicalize(value)
        elif isinstance(value, bytes):
            return dns.ipv6.inet_ntoa(value)
        else:
            raise ValueError("not an IPv6 address")

@classmethod
    def _as_bool(cls, value):
        if isinstance(value, bool):
            return value
        else:
            raise ValueError("not a boolean")

@classmethod
    def _as_ttl(cls, value):
        if isinstance(value, int):
            return cls._as_int(value, 0, dns.ttl.MAX_TTL)
        elif isinstance(value, str):
            return dns.ttl.from_text(value)
        else:
            raise ValueError("not a TTL")

@classmethod
    def _as_tuple(cls, value, as_value):
        try:
            # For user convenience, if value is a singleton of the list
            # element type, wrap it in a tuple.
            return (as_value(value),)
        except Exception:
            # Otherwise, check each element of the iterable *value*
            # against *as_value*.
            return tuple(as_value(v) for v in value)

# Processing order

@classmethod
    def _processing_order(cls, iterable):
        items = list(iterable)
        random.shuffle(items)
        return items

@dns.immutable.immutable
class GenericRdata(Rdata):
    """Generic Rdata Class

This class is used for rdata types for which we have no better
    implementation.  It implements the DNS "unknown RRs" scheme.
    """

__slots__ = ["data"]

def __init__(self, rdclass, rdtype, data):
        super().__init__(rdclass, rdtype)
        self.data = data

def to_text(
        self,
        origin: Optional[dns.name.Name] = None,
        relativize: bool = True,
        **kw: Dict[str, Any],
    ) -> str:
        return r"\# %d " % len(self.data) + _hexify(self.data, **kw)

@classmethod
    def from_text(
        cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None
    ):
        token = tok.get()
        if not token.is_identifier() or token.value != r"\#":
            raise dns.exception.SyntaxError(r"generic rdata does not start with \#")
        length = tok.get_int()
        hex = tok.concatenate_remaining_identifiers(True).encode()
        data = binascii.unhexlify(hex)
        if len(data) != length:
            raise dns.exception.SyntaxError("generic rdata hex data has wrong length")
        return cls(rdclass, rdtype, data)

def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
        file.write(self.data)

@classmethod
    def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
        return cls(rdclass, rdtype, parser.get_remaining())

_rdata_classes: Dict[Tuple[dns.rdataclass.RdataClass, dns.rdatatype.RdataType], Any] = (
    {}
)
_module_prefix = "dns.rdtypes"
_dynamic_load_allowed = True

def get_rdata_class(rdclass, rdtype, use_generic=True):
    cls = _rdata_classes.get((rdclass, rdtype))
    if not cls:
        cls = _rdata_classes.get((dns.rdatatype.ANY, rdtype))
        if not cls and _dynamic_load_allowed:
            rdclass_text = dns.rdataclass.to_text(rdclass)
            rdtype_text = dns.rdatatype.to_text(rdtype)
            rdtype_text = rdtype_text.replace("-", "_")
            try:
                mod = import_module(
                    ".".join([_module_prefix, rdclass_text, rdtype_text])
                )
                cls = getattr(mod, rdtype_text)
                _rdata_classes[(rdclass, rdtype)] = cls
            except ImportError:
                try:
                    mod = import_module(".".join([_module_prefix, "ANY", rdtype_text]))
                    cls = getattr(mod, rdtype_text)
                    _rdata_classes[(dns.rdataclass.ANY, rdtype)] = cls
                    _rdata_classes[(rdclass, rdtype)] = cls
                except ImportError:
                    pass
    if not cls and use_generic:
        cls = GenericRdata
        _rdata_classes[(rdclass, rdtype)] = cls
    return cls

def load_all_types(disable_dynamic_load=True):
    """Load all rdata types for which dnspython has a non-generic implementation.

Normally dnspython loads DNS rdatatype implementations on demand, but in some
    specialized cases loading all types at an application-controlled time is preferred.

If *disable_dynamic_load*, a ``bool``, is ``True`` then dnspython will not attempt
    to use its dynamic loading mechanism if an unknown type is subsequently encountered,
    and will simply use the ``GenericRdata`` class.
    """
    # Load class IN and ANY types.
    for rdtype in dns.rdatatype.RdataType:
        get_rdata_class(dns.rdataclass.IN, rdtype, False)
    # Load the one non-ANY implementation we have in CH.  Everything
    # else in CH is an ANY type, and we'll discover those on demand but won't
    # have to import anything.
    get_rdata_class(dns.rdataclass.CH, dns.rdatatype.A, False)
    if disable_dynamic_load:
        # Now disable dynamic loading so any subsequent unknown type immediately becomes
        # GenericRdata without a load attempt.
        global _dynamic_load_allowed
        _dynamic_load_allowed = False

def from_text(
    rdclass: Union[dns.rdataclass.RdataClass, str],
    rdtype: Union[dns.rdatatype.RdataType, str],
    tok: Union[dns.tokenizer.Tokenizer, str],
    origin: Optional[dns.name.Name] = None,
    relativize: bool = True,
    relativize_to: Optional[dns.name.Name] = None,
    idna_codec: Optional[dns.name.IDNACodec] = None,
) -> Rdata:
    """Build an rdata object from text format.

This function attempts to dynamically load a class which
    implements the specified rdata class and type.  If there is no
    class-and-type-specific implementation, the GenericRdata class
    is used.

Once a class is chosen, its from_text() class method is called
    with the parameters to this function.

If *tok* is a ``str``, then a tokenizer is created and the string
    is used as its input.

*rdclass*, a ``dns.rdataclass.RdataClass`` or ``str``, the rdataclass.

*rdtype*, a ``dns.rdatatype.RdataType`` or ``str``, the rdatatype.

*tok*, a ``dns.tokenizer.Tokenizer`` or a ``str``.

*origin*, a ``dns.name.Name`` (or ``None``), the
    origin to use for relative names.

*relativize*, a ``bool``.  If true, name will be relativized.

*relativize_to*, a ``dns.name.Name`` (or ``None``), the origin to use
    when relativizing names.  If not set, the *origin* value will be used.

*idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA
    encoder/decoder to use if a tokenizer needs to be created.  If
    ``None``, the default IDNA 2003 encoder/decoder is used.  If a
    tokenizer is not created, then the codec associated with the tokenizer
    is the one that is used.

Returns an instance of the chosen Rdata subclass.

"""
    if isinstance(tok, str):
        tok = dns.tokenizer.Tokenizer(tok, idna_codec=idna_codec)
    rdclass = dns.rdataclass.RdataClass.make(rdclass)
    rdtype = dns.rdatatype.RdataType.make(rdtype)
    cls = get_rdata_class(rdclass, rdtype)
    with dns.exception.ExceptionWrapper(dns.exception.SyntaxError):
        rdata = None
        if cls != GenericRdata:
            # peek at first token
            token = tok.get()
            tok.unget(token)
            if token.is_identifier() and token.value == r"\#":
                #
                # Known type using the generic syntax.  Extract the
                # wire form from the generic syntax, and then run
                # from_wire on it.
                #
                grdata = GenericRdata.from_text(
                    rdclass, rdtype, tok, origin, relativize, relativize_to
                )
                rdata = from_wire(
                    rdclass, rdtype, grdata.data, 0, len(grdata.data), origin
                )
                #
                # If this comparison isn't equal, then there must have been
                # compressed names in the wire format, which is an error,
                # there being no reasonable context to decompress with.
                #
                rwire = rdata.to_wire()
                if rwire != grdata.data:
                    raise dns.exception.SyntaxError(
                        "compressed data in "
                        "generic syntax form "
                        "of known rdatatype"
                    )
        if rdata is None:
            rdata = cls.from_text(
                rdclass, rdtype, tok, origin, relativize, relativize_to
            )
        token = tok.get_eol_as_token()
        if token.comment is not None:
            object.__setattr__(rdata, "rdcomment", token.comment)
        return rdata

def from_wire_parser(
    rdclass: Union[dns.rdataclass.RdataClass, str],
    rdtype: Union[dns.rdatatype.RdataType, str],
    parser: dns.wire.Parser,
    origin: Optional[dns.name.Name] = None,
) -> Rdata:
    """Build an rdata object from wire format

Once a class is chosen, its from_wire() class method is called
    with the parameters to this function.

*rdclass*, a ``dns.rdataclass.RdataClass`` or ``str``, the rdataclass.

*rdtype*, a ``dns.rdatatype.RdataType`` or ``str``, the rdatatype.

*parser*, a ``dns.wire.Parser``, the parser, which should be
    restricted to the rdata length.

*origin*, a ``dns.name.Name`` (or ``None``).  If not ``None``,
    then names will be relativized to this origin.

Returns an instance of the chosen Rdata subclass.
    """

rdclass = dns.rdataclass.RdataClass.make(rdclass)
    rdtype = dns.rdatatype.RdataType.make(rdtype)
    cls = get_rdata_class(rdclass, rdtype)
    with dns.exception.ExceptionWrapper(dns.exception.FormError):
        return cls.from_wire_parser(rdclass, rdtype, parser, origin)

def from_wire(
    rdclass: Union[dns.rdataclass.RdataClass, str],
    rdtype: Union[dns.rdatatype.RdataType, str],
    wire: bytes,
    current: int,
    rdlen: int,
    origin: Optional[dns.name.Name] = None,
) -> Rdata:
    """Build an rdata object from wire format

Once a class is chosen, its from_wire() class method is called
    with the parameters to this function.

*rdclass*, an ``int``, the rdataclass.

*rdtype*, an ``int``, the rdatatype.

*wire*, a ``bytes``, the wire-format message.

*current*, an ``int``, the offset in wire of the beginning of
    the rdata.

*rdlen*, an ``int``, the length of the wire-format rdata

*origin*, a ``dns.name.Name`` (or ``None``).  If not ``None``,
    then names will be relativized to this origin.

Returns an instance of the chosen Rdata subclass.
    """
    parser = dns.wire.Parser(wire, current)
    with parser.restrict_to(rdlen):
        return from_wire_parser(rdclass, rdtype, parser, origin)

class RdatatypeExists(dns.exception.DNSException):
    """DNS rdatatype already exists."""

supp_kwargs = {"rdclass", "rdtype"}
    fmt = (
        "The rdata type with class {rdclass:d} and rdtype {rdtype:d} "
        + "already exists."
    )

def register_type(
    implementation: Any,
    rdtype: int,
    rdtype_text: str,
    is_singleton: bool = False,
    rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
) -> None:
    """Dynamically register a module to handle an rdatatype.

*implementation*, a module implementing the type in the usual dnspython
    way.

*rdtype*, an ``int``, the rdatatype to register.

*rdtype_text*, a ``str``, the textual form of the rdatatype.

*is_singleton*, a ``bool``, indicating if the type is a singleton (i.e.
    RRsets of the type can have only one member.)

*rdclass*, the rdataclass of the type, or ``dns.rdataclass.ANY`` if
    it applies to all classes.
    """

rdtype = dns.rdatatype.RdataType.make(rdtype)
    existing_cls = get_rdata_class(rdclass, rdtype)
    if existing_cls != GenericRdata or dns.rdatatype.is_metatype(rdtype):
        raise RdatatypeExists(rdclass=rdclass, rdtype=rdtype)
    _rdata_classes[(rdclass, rdtype)] = getattr(
        implementation, rdtype_text.replace("-", "_")
    )
    dns.rdatatype.register_type(rdtype, rdtype_text, is_singleton)

Back