Загрузить файлы в «venv/Lib/site-packages/dns»

This commit is contained in:
2026-07-02 18:07:30 +00:00
parent 8fc1d456ba
commit 2b62519d62
5 changed files with 2067 additions and 0 deletions

View File

@@ -0,0 +1,168 @@
# 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 Result Codes."""
from typing import Tuple, Type
import dns.enum
import dns.exception
class Rcode(dns.enum.IntEnum):
#: No error
NOERROR = 0
#: Format error
FORMERR = 1
#: Server failure
SERVFAIL = 2
#: Name does not exist ("Name Error" in RFC 1025 terminology).
NXDOMAIN = 3
#: Not implemented
NOTIMP = 4
#: Refused
REFUSED = 5
#: Name exists.
YXDOMAIN = 6
#: RRset exists.
YXRRSET = 7
#: RRset does not exist.
NXRRSET = 8
#: Not authoritative.
NOTAUTH = 9
#: Name not in zone.
NOTZONE = 10
#: DSO-TYPE Not Implemented
DSOTYPENI = 11
#: Bad EDNS version.
BADVERS = 16
#: TSIG Signature Failure
BADSIG = 16
#: Key not recognized.
BADKEY = 17
#: Signature out of time window.
BADTIME = 18
#: Bad TKEY Mode.
BADMODE = 19
#: Duplicate key name.
BADNAME = 20
#: Algorithm not supported.
BADALG = 21
#: Bad Truncation
BADTRUNC = 22
#: Bad/missing Server Cookie
BADCOOKIE = 23
@classmethod
def _maximum(cls):
return 4095
@classmethod
def _unknown_exception_class(cls) -> Type[Exception]:
return UnknownRcode
class UnknownRcode(dns.exception.DNSException):
"""A DNS rcode is unknown."""
def from_text(text: str) -> Rcode:
"""Convert text into an rcode.
*text*, a ``str``, the textual rcode or an integer in textual form.
Raises ``dns.rcode.UnknownRcode`` if the rcode mnemonic is unknown.
Returns a ``dns.rcode.Rcode``.
"""
return Rcode.from_text(text)
def from_flags(flags: int, ednsflags: int) -> Rcode:
"""Return the rcode value encoded by flags and ednsflags.
*flags*, an ``int``, the DNS flags field.
*ednsflags*, an ``int``, the EDNS flags field.
Raises ``ValueError`` if rcode is < 0 or > 4095
Returns a ``dns.rcode.Rcode``.
"""
value = (flags & 0x000F) | ((ednsflags >> 20) & 0xFF0)
return Rcode.make(value)
def to_flags(value: Rcode) -> Tuple[int, int]:
"""Return a (flags, ednsflags) tuple which encodes the rcode.
*value*, a ``dns.rcode.Rcode``, the rcode.
Raises ``ValueError`` if rcode is < 0 or > 4095.
Returns an ``(int, int)`` tuple.
"""
if value < 0 or value > 4095:
raise ValueError("rcode must be >= 0 and <= 4095")
v = value & 0xF
ev = (value & 0xFF0) << 20
return (v, ev)
def to_text(value: Rcode, tsig: bool = False) -> str:
"""Convert rcode into text.
*value*, a ``dns.rcode.Rcode``, the rcode.
Raises ``ValueError`` if rcode is < 0 or > 4095.
Returns a ``str``.
"""
if tsig and value == Rcode.BADVERS:
return "BADSIG"
return Rcode.to_text(value)
### BEGIN generated Rcode constants
NOERROR = Rcode.NOERROR
FORMERR = Rcode.FORMERR
SERVFAIL = Rcode.SERVFAIL
NXDOMAIN = Rcode.NXDOMAIN
NOTIMP = Rcode.NOTIMP
REFUSED = Rcode.REFUSED
YXDOMAIN = Rcode.YXDOMAIN
YXRRSET = Rcode.YXRRSET
NXRRSET = Rcode.NXRRSET
NOTAUTH = Rcode.NOTAUTH
NOTZONE = Rcode.NOTZONE
DSOTYPENI = Rcode.DSOTYPENI
BADVERS = Rcode.BADVERS
BADSIG = Rcode.BADSIG
BADKEY = Rcode.BADKEY
BADTIME = Rcode.BADTIME
BADMODE = Rcode.BADMODE
BADNAME = Rcode.BADNAME
BADALG = Rcode.BADALG
BADTRUNC = Rcode.BADTRUNC
BADCOOKIE = Rcode.BADCOOKIE
### END generated Rcode constants

View File

@@ -0,0 +1,935 @@
# 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 ipaddress
import itertools
import random
from importlib import import_module
from typing import Any, Dict, Tuple
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 += f"\\{c:03d}"
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: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
) -> None:
"""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: dns.name.Name | None = 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: Any,
compress: dns.name.CompressType | None = None,
origin: dns.name.Name | None = None,
canonicalize: bool = False,
) -> None:
raise NotImplementedError # pragma: no cover
def to_wire(
self,
file: Any | None = None,
compress: dns.name.CompressType | None = None,
origin: dns.name.Name | None = None,
canonicalize: bool = False,
) -> bytes | None:
"""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: dns.name.Name | None = None) -> "GenericRdata":
"""Creates a dns.rdata.GenericRdata equivalent of this rdata.
Returns a ``dns.rdata.GenericRdata``.
"""
wire = self.to_wire(origin=origin)
assert wire is not None # for type checkers
return GenericRdata(self.rdclass, self.rdtype, wire)
def to_digestable(self, origin: dns.name.Name | None = 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.
"""
# the next two lines are for type checkers, so they are bound
our = b""
their = b""
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: dns.name.Name | None = None,
relativize: bool = True,
relativize_to: dns.name.Name | None = 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: dns.name.Name | None = 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: int | None = 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)
elif isinstance(value, ipaddress.IPv4Address):
return dns.ipv4.inet_ntoa(value.packed)
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)
elif isinstance(value, ipaddress.IPv6Address):
return dns.ipv6.inet_ntoa(value.packed)
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: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
data: bytes,
) -> None:
super().__init__(rdclass, rdtype)
self.data = data
def to_text(
self,
origin: dns.name.Name | None = None,
relativize: bool = True,
**kw: Dict[str, Any],
) -> str:
return rf"\# {len(self.data)} " + _hexify(self.data, **kw) # pyright: ignore
@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)
def to_generic(self, origin: dns.name.Name | None = None) -> "GenericRdata":
return self
@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.rdataclass.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: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
tok: dns.tokenizer.Tokenizer | str,
origin: dns.name.Name | None = None,
relativize: bool = True,
relativize_to: dns.name.Name | None = None,
idna_codec: dns.name.IDNACodec | None = 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)
if not isinstance(tok, dns.tokenizer.Tokenizer):
raise ValueError("tok must be a string or a Tokenizer")
rdclass = dns.rdataclass.RdataClass.make(rdclass)
rdtype = dns.rdatatype.RdataType.make(rdtype)
cls = get_rdata_class(rdclass, rdtype)
assert cls is not None # for type checkers
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: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
parser: dns.wire.Parser,
origin: dns.name.Name | None = 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)
assert cls is not None # for type checkers
with dns.exception.ExceptionWrapper(dns.exception.FormError):
return cls.from_wire_parser(rdclass, rdtype, parser, origin)
def from_wire(
rdclass: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
wire: bytes,
current: int,
rdlen: int,
origin: dns.name.Name | None = 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 subclass of ``dns.rdata.Rdata`` implementing the type,
or a module containing such a class named by its text form.
*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)
if isinstance(implementation, type) and issubclass(implementation, Rdata):
impclass = implementation
else:
impclass = getattr(implementation, rdtype_text.replace("-", "_"))
_rdata_classes[(rdclass, rdtype)] = impclass
dns.rdatatype.register_type(rdtype, rdtype_text, is_singleton)

View File

@@ -0,0 +1,118 @@
# 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 Classes."""
import dns.enum
import dns.exception
class RdataClass(dns.enum.IntEnum):
"""DNS Rdata Class"""
RESERVED0 = 0
IN = 1
INTERNET = IN
CH = 3
CHAOS = CH
HS = 4
HESIOD = HS
NONE = 254
ANY = 255
@classmethod
def _maximum(cls):
return 65535
@classmethod
def _short_name(cls):
return "class"
@classmethod
def _prefix(cls):
return "CLASS"
@classmethod
def _unknown_exception_class(cls):
return UnknownRdataclass
_metaclasses = {RdataClass.NONE, RdataClass.ANY}
class UnknownRdataclass(dns.exception.DNSException):
"""A DNS class is unknown."""
def from_text(text: str) -> RdataClass:
"""Convert text into a DNS rdata class value.
The input text can be a defined DNS RR class mnemonic or
instance of the DNS generic class syntax.
For example, "IN" and "CLASS1" will both result in a value of 1.
Raises ``dns.rdatatype.UnknownRdataclass`` if the class is unknown.
Raises ``ValueError`` if the rdata class value is not >= 0 and <= 65535.
Returns a ``dns.rdataclass.RdataClass``.
"""
return RdataClass.from_text(text)
def to_text(value: RdataClass) -> str:
"""Convert a DNS rdata class value to text.
If the value has a known mnemonic, it will be used, otherwise the
DNS generic class syntax will be used.
Raises ``ValueError`` if the rdata class value is not >= 0 and <= 65535.
Returns a ``str``.
"""
return RdataClass.to_text(value)
def is_metaclass(rdclass: RdataClass) -> bool:
"""True if the specified class is a metaclass.
The currently defined metaclasses are ANY and NONE.
*rdclass* is a ``dns.rdataclass.RdataClass``.
"""
if rdclass in _metaclasses:
return True
return False
### BEGIN generated RdataClass constants
RESERVED0 = RdataClass.RESERVED0
IN = RdataClass.IN
INTERNET = RdataClass.INTERNET
CH = RdataClass.CH
CHAOS = RdataClass.CHAOS
HS = RdataClass.HS
HESIOD = RdataClass.HESIOD
NONE = RdataClass.NONE
ANY = RdataClass.ANY
### END generated RdataClass constants

View File

@@ -0,0 +1,508 @@
# 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 rdatasets (an rdataset is a set of rdatas of a given type and class)"""
import io
import random
import struct
from typing import Any, Collection, Dict, List, cast
import dns.exception
import dns.immutable
import dns.name
import dns.rdata
import dns.rdataclass
import dns.rdatatype
import dns.renderer
import dns.set
import dns.ttl
# define SimpleSet here for backwards compatibility
SimpleSet = dns.set.Set
class DifferingCovers(dns.exception.DNSException):
"""An attempt was made to add a DNS SIG/RRSIG whose covered type
is not the same as that of the other rdatas in the rdataset."""
class IncompatibleTypes(dns.exception.DNSException):
"""An attempt was made to add DNS RR data of an incompatible type."""
class Rdataset(dns.set.Set):
"""A DNS rdataset."""
__slots__ = ["rdclass", "rdtype", "covers", "ttl"]
def __init__(
self,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
ttl: int = 0,
):
"""Create a new rdataset of the specified class and type.
*rdclass*, a ``dns.rdataclass.RdataClass``, the rdataclass.
*rdtype*, an ``dns.rdatatype.RdataType``, the rdatatype.
*covers*, an ``dns.rdatatype.RdataType``, the covered rdatatype.
*ttl*, an ``int``, the TTL.
"""
super().__init__()
self.rdclass = rdclass
self.rdtype: dns.rdatatype.RdataType = rdtype
self.covers: dns.rdatatype.RdataType = covers
self.ttl = ttl
def _clone(self):
obj = cast(Rdataset, super()._clone())
obj.rdclass = self.rdclass
obj.rdtype = self.rdtype
obj.covers = self.covers
obj.ttl = self.ttl
return obj
def update_ttl(self, ttl: int) -> None:
"""Perform TTL minimization.
Set the TTL of the rdataset to be the lesser of the set's current
TTL or the specified TTL. If the set contains no rdatas, set the TTL
to the specified TTL.
*ttl*, an ``int`` or ``str``.
"""
ttl = dns.ttl.make(ttl)
if len(self) == 0:
self.ttl = ttl
elif ttl < self.ttl:
self.ttl = ttl
# pylint: disable=arguments-differ,arguments-renamed
def add( # pyright: ignore
self, rd: dns.rdata.Rdata, ttl: int | None = None
) -> None:
"""Add the specified rdata to the rdataset.
If the optional *ttl* parameter is supplied, then
``self.update_ttl(ttl)`` will be called prior to adding the rdata.
*rd*, a ``dns.rdata.Rdata``, the rdata
*ttl*, an ``int``, the TTL.
Raises ``dns.rdataset.IncompatibleTypes`` if the type and class
do not match the type and class of the rdataset.
Raises ``dns.rdataset.DifferingCovers`` if the type is a signature
type and the covered type does not match that of the rdataset.
"""
#
# If we're adding a signature, do some special handling to
# check that the signature covers the same type as the
# other rdatas in this rdataset. If this is the first rdata
# in the set, initialize the covers field.
#
if self.rdclass != rd.rdclass or self.rdtype != rd.rdtype:
raise IncompatibleTypes
if ttl is not None:
self.update_ttl(ttl)
if self.rdtype == dns.rdatatype.RRSIG or self.rdtype == dns.rdatatype.SIG:
covers = rd.covers()
if len(self) == 0 and self.covers == dns.rdatatype.NONE:
self.covers = covers
elif self.covers != covers:
raise DifferingCovers
if dns.rdatatype.is_singleton(rd.rdtype) and len(self) > 0:
self.clear()
super().add(rd)
def union_update(self, other):
self.update_ttl(other.ttl)
super().union_update(other)
def intersection_update(self, other):
self.update_ttl(other.ttl)
super().intersection_update(other)
def update(self, other):
"""Add all rdatas in other to self.
*other*, a ``dns.rdataset.Rdataset``, the rdataset from which
to update.
"""
self.update_ttl(other.ttl)
super().update(other)
def _rdata_repr(self):
def maybe_truncate(s):
if len(s) > 100:
return s[:100] + "..."
return s
return "[" + ", ".join(f"<{maybe_truncate(str(rr))}>" for rr in self) + "]"
def __repr__(self):
if self.covers == 0:
ctext = ""
else:
ctext = "(" + dns.rdatatype.to_text(self.covers) + ")"
return (
"<DNS "
+ dns.rdataclass.to_text(self.rdclass)
+ " "
+ dns.rdatatype.to_text(self.rdtype)
+ ctext
+ " rdataset: "
+ self._rdata_repr()
+ ">"
)
def __str__(self):
return self.to_text()
def __eq__(self, other):
if not isinstance(other, Rdataset):
return False
if (
self.rdclass != other.rdclass
or self.rdtype != other.rdtype
or self.covers != other.covers
):
return False
return super().__eq__(other)
def __ne__(self, other):
return not self.__eq__(other)
def to_text(
self,
name: dns.name.Name | None = None,
origin: dns.name.Name | None = None,
relativize: bool = True,
override_rdclass: dns.rdataclass.RdataClass | None = None,
want_comments: bool = False,
**kw: Dict[str, Any],
) -> str:
"""Convert the rdataset into DNS zone file format.
See ``dns.name.Name.choose_relativity`` for more information
on how *origin* and *relativize* determine the way names
are emitted.
Any additional keyword arguments are passed on to the rdata
``to_text()`` method.
*name*, a ``dns.name.Name``. If name is not ``None``, emit RRs with
*name* as the owner name.
*origin*, a ``dns.name.Name`` or ``None``, the origin for relative
names.
*relativize*, a ``bool``. If ``True``, names will be relativized
to *origin*.
*override_rdclass*, a ``dns.rdataclass.RdataClass`` or ``None``.
If not ``None``, use this class instead of the Rdataset's class.
*want_comments*, a ``bool``. If ``True``, emit comments for rdata
which have them. The default is ``False``.
"""
if name is not None:
name = name.choose_relativity(origin, relativize)
ntext = str(name)
pad = " "
else:
ntext = ""
pad = ""
s = io.StringIO()
if override_rdclass is not None:
rdclass = override_rdclass
else:
rdclass = self.rdclass
if len(self) == 0:
#
# Empty rdatasets are used for the question section, and in
# some dynamic updates, so we don't need to print out the TTL
# (which is meaningless anyway).
#
s.write(
f"{ntext}{pad}{dns.rdataclass.to_text(rdclass)} "
f"{dns.rdatatype.to_text(self.rdtype)}\n"
)
else:
for rd in self:
extra = ""
if want_comments:
if rd.rdcomment:
extra = f" ;{rd.rdcomment}"
s.write(
f"{ntext}{pad}{self.ttl} "
f"{dns.rdataclass.to_text(rdclass)} "
f"{dns.rdatatype.to_text(self.rdtype)} "
f"{rd.to_text(origin=origin, relativize=relativize, **kw)}"
f"{extra}\n"
)
#
# We strip off the final \n for the caller's convenience in printing
#
return s.getvalue()[:-1]
def to_wire(
self,
name: dns.name.Name,
file: Any,
compress: dns.name.CompressType | None = None,
origin: dns.name.Name | None = None,
override_rdclass: dns.rdataclass.RdataClass | None = None,
want_shuffle: bool = True,
) -> int:
"""Convert the rdataset to wire format.
*name*, a ``dns.name.Name`` is the owner name to use.
*file* is the file where the name is emitted (typically a
BytesIO file).
*compress*, a ``dict``, is the compression table to use. If
``None`` (the default), names will not be compressed.
*origin* is a ``dns.name.Name`` or ``None``. If the name is
relative and origin is not ``None``, then *origin* will be appended
to it.
*override_rdclass*, an ``int``, is used as the class instead of the
class of the rdataset. This is useful when rendering rdatasets
associated with dynamic updates.
*want_shuffle*, a ``bool``. If ``True``, then the order of the
Rdatas within the Rdataset will be shuffled before rendering.
Returns an ``int``, the number of records emitted.
"""
if override_rdclass is not None:
rdclass = override_rdclass
want_shuffle = False
else:
rdclass = self.rdclass
if len(self) == 0:
name.to_wire(file, compress, origin)
file.write(struct.pack("!HHIH", self.rdtype, rdclass, 0, 0))
return 1
else:
l: Rdataset | List[dns.rdata.Rdata]
if want_shuffle:
l = list(self)
random.shuffle(l)
else:
l = self
for rd in l:
name.to_wire(file, compress, origin)
file.write(struct.pack("!HHI", self.rdtype, rdclass, self.ttl))
with dns.renderer.prefixed_length(file, 2):
rd.to_wire(file, compress, origin)
return len(self)
def match(
self,
rdclass: dns.rdataclass.RdataClass,
rdtype: dns.rdatatype.RdataType,
covers: dns.rdatatype.RdataType,
) -> bool:
"""Returns ``True`` if this rdataset matches the specified class,
type, and covers.
"""
if self.rdclass == rdclass and self.rdtype == rdtype and self.covers == covers:
return True
return False
def processing_order(self) -> List[dns.rdata.Rdata]:
"""Return rdatas in a valid processing order according to the type's
specification. For example, MX records are in preference order from
lowest to highest preferences, with items of the same preference
shuffled.
For types that do not define a processing order, the rdatas are
simply shuffled.
"""
if len(self) == 0:
return []
else:
return self[0]._processing_order(iter(self)) # pyright: ignore
@dns.immutable.immutable
class ImmutableRdataset(Rdataset): # lgtm[py/missing-equals]
"""An immutable DNS rdataset."""
_clone_class = Rdataset
def __init__(self, rdataset: Rdataset):
"""Create an immutable rdataset from the specified rdataset."""
super().__init__(
rdataset.rdclass, rdataset.rdtype, rdataset.covers, rdataset.ttl
)
self.items = dns.immutable.Dict(rdataset.items)
def update_ttl(self, ttl):
raise TypeError("immutable")
def add(self, rd, ttl=None):
raise TypeError("immutable")
def union_update(self, other):
raise TypeError("immutable")
def intersection_update(self, other):
raise TypeError("immutable")
def update(self, other):
raise TypeError("immutable")
def __delitem__(self, i):
raise TypeError("immutable")
# lgtm complains about these not raising ArithmeticError, but there is
# precedent for overrides of these methods in other classes to raise
# TypeError, and it seems like the better exception.
def __ior__(self, other): # lgtm[py/unexpected-raise-in-special-method]
raise TypeError("immutable")
def __iand__(self, other): # lgtm[py/unexpected-raise-in-special-method]
raise TypeError("immutable")
def __iadd__(self, other): # lgtm[py/unexpected-raise-in-special-method]
raise TypeError("immutable")
def __isub__(self, other): # lgtm[py/unexpected-raise-in-special-method]
raise TypeError("immutable")
def clear(self):
raise TypeError("immutable")
def __copy__(self):
return ImmutableRdataset(super().copy()) # pyright: ignore
def copy(self):
return ImmutableRdataset(super().copy()) # pyright: ignore
def union(self, other):
return ImmutableRdataset(super().union(other)) # pyright: ignore
def intersection(self, other):
return ImmutableRdataset(super().intersection(other)) # pyright: ignore
def difference(self, other):
return ImmutableRdataset(super().difference(other)) # pyright: ignore
def symmetric_difference(self, other):
return ImmutableRdataset(super().symmetric_difference(other)) # pyright: ignore
def from_text_list(
rdclass: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
ttl: int,
text_rdatas: Collection[str],
idna_codec: dns.name.IDNACodec | None = None,
origin: dns.name.Name | None = None,
relativize: bool = True,
relativize_to: dns.name.Name | None = None,
) -> Rdataset:
"""Create an rdataset with the specified class, type, and TTL, and with
the specified list of rdatas in text format.
*idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA
encoder/decoder to use; if ``None``, the default IDNA 2003
encoder/decoder is used.
*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.
Returns a ``dns.rdataset.Rdataset`` object.
"""
rdclass = dns.rdataclass.RdataClass.make(rdclass)
rdtype = dns.rdatatype.RdataType.make(rdtype)
r = Rdataset(rdclass, rdtype)
r.update_ttl(ttl)
for t in text_rdatas:
rd = dns.rdata.from_text(
r.rdclass, r.rdtype, t, origin, relativize, relativize_to, idna_codec
)
r.add(rd)
return r
def from_text(
rdclass: dns.rdataclass.RdataClass | str,
rdtype: dns.rdatatype.RdataType | str,
ttl: int,
*text_rdatas: Any,
) -> Rdataset:
"""Create an rdataset with the specified class, type, and TTL, and with
the specified rdatas in text format.
Returns a ``dns.rdataset.Rdataset`` object.
"""
return from_text_list(rdclass, rdtype, ttl, cast(Collection[str], text_rdatas))
def from_rdata_list(ttl: int, rdatas: Collection[dns.rdata.Rdata]) -> Rdataset:
"""Create an rdataset with the specified TTL, and with
the specified list of rdata objects.
Returns a ``dns.rdataset.Rdataset`` object.
"""
if len(rdatas) == 0:
raise ValueError("rdata list must not be empty")
r = None
for rd in rdatas:
if r is None:
r = Rdataset(rd.rdclass, rd.rdtype)
r.update_ttl(ttl)
r.add(rd)
assert r is not None
return r
def from_rdata(ttl: int, *rdatas: Any) -> Rdataset:
"""Create an rdataset with the specified TTL, and with
the specified rdata objects.
Returns a ``dns.rdataset.Rdataset`` object.
"""
return from_rdata_list(ttl, cast(Collection[dns.rdata.Rdata], rdatas))

View File

@@ -0,0 +1,338 @@
# 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 Types."""
from typing import Dict
import dns.enum
import dns.exception
class RdataType(dns.enum.IntEnum):
"""DNS Rdata Type"""
TYPE0 = 0
NONE = 0
A = 1
NS = 2
MD = 3
MF = 4
CNAME = 5
SOA = 6
MB = 7
MG = 8
MR = 9
NULL = 10
WKS = 11
PTR = 12
HINFO = 13
MINFO = 14
MX = 15
TXT = 16
RP = 17
AFSDB = 18
X25 = 19
ISDN = 20
RT = 21
NSAP = 22
NSAP_PTR = 23
SIG = 24
KEY = 25
PX = 26
GPOS = 27
AAAA = 28
LOC = 29
NXT = 30
SRV = 33
NAPTR = 35
KX = 36
CERT = 37
A6 = 38
DNAME = 39
OPT = 41
APL = 42
DS = 43
SSHFP = 44
IPSECKEY = 45
RRSIG = 46
NSEC = 47
DNSKEY = 48
DHCID = 49
NSEC3 = 50
NSEC3PARAM = 51
TLSA = 52
SMIMEA = 53
HIP = 55
NINFO = 56
CDS = 59
CDNSKEY = 60
OPENPGPKEY = 61
CSYNC = 62
ZONEMD = 63
SVCB = 64
HTTPS = 65
DSYNC = 66
SPF = 99
UNSPEC = 103
NID = 104
L32 = 105
L64 = 106
LP = 107
EUI48 = 108
EUI64 = 109
TKEY = 249
TSIG = 250
IXFR = 251
AXFR = 252
MAILB = 253
MAILA = 254
ANY = 255
URI = 256
CAA = 257
AVC = 258
AMTRELAY = 260
RESINFO = 261
WALLET = 262
TA = 32768
DLV = 32769
@classmethod
def _maximum(cls):
return 65535
@classmethod
def _short_name(cls):
return "type"
@classmethod
def _prefix(cls):
return "TYPE"
@classmethod
def _extra_from_text(cls, text):
if text.find("-") >= 0:
try:
return cls[text.replace("-", "_")]
except KeyError: # pragma: no cover
pass
return _registered_by_text.get(text)
@classmethod
def _extra_to_text(cls, value, current_text):
if current_text is None:
return _registered_by_value.get(value)
if current_text.find("_") >= 0:
return current_text.replace("_", "-")
return current_text
@classmethod
def _unknown_exception_class(cls):
return UnknownRdatatype
_registered_by_text: Dict[str, RdataType] = {}
_registered_by_value: Dict[RdataType, str] = {}
_metatypes = {RdataType.OPT}
_singletons = {
RdataType.SOA,
RdataType.NXT,
RdataType.DNAME,
RdataType.NSEC,
RdataType.CNAME,
}
class UnknownRdatatype(dns.exception.DNSException):
"""DNS resource record type is unknown."""
def from_text(text: str) -> RdataType:
"""Convert text into a DNS rdata type value.
The input text can be a defined DNS RR type mnemonic or
instance of the DNS generic type syntax.
For example, "NS" and "TYPE2" will both result in a value of 2.
Raises ``dns.rdatatype.UnknownRdatatype`` if the type is unknown.
Raises ``ValueError`` if the rdata type value is not >= 0 and <= 65535.
Returns a ``dns.rdatatype.RdataType``.
"""
return RdataType.from_text(text)
def to_text(value: RdataType) -> str:
"""Convert a DNS rdata type value to text.
If the value has a known mnemonic, it will be used, otherwise the
DNS generic type syntax will be used.
Raises ``ValueError`` if the rdata type value is not >= 0 and <= 65535.
Returns a ``str``.
"""
return RdataType.to_text(value)
def is_metatype(rdtype: RdataType) -> bool:
"""True if the specified type is a metatype.
*rdtype* is a ``dns.rdatatype.RdataType``.
The currently defined metatypes are TKEY, TSIG, IXFR, AXFR, MAILA,
MAILB, ANY, and OPT.
Returns a ``bool``.
"""
return (256 > rdtype >= 128) or rdtype in _metatypes
def is_singleton(rdtype: RdataType) -> bool:
"""Is the specified type a singleton type?
Singleton types can only have a single rdata in an rdataset, or a single
RR in an RRset.
The currently defined singleton types are CNAME, DNAME, NSEC, NXT, and
SOA.
*rdtype* is an ``int``.
Returns a ``bool``.
"""
if rdtype in _singletons:
return True
return False
# pylint: disable=redefined-outer-name
def register_type(
rdtype: RdataType, rdtype_text: str, is_singleton: bool = False
) -> None:
"""Dynamically register an rdatatype.
*rdtype*, a ``dns.rdatatype.RdataType``, 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.)
"""
_registered_by_text[rdtype_text] = rdtype
_registered_by_value[rdtype] = rdtype_text
if is_singleton:
_singletons.add(rdtype)
### BEGIN generated RdataType constants
TYPE0 = RdataType.TYPE0
NONE = RdataType.NONE
A = RdataType.A
NS = RdataType.NS
MD = RdataType.MD
MF = RdataType.MF
CNAME = RdataType.CNAME
SOA = RdataType.SOA
MB = RdataType.MB
MG = RdataType.MG
MR = RdataType.MR
NULL = RdataType.NULL
WKS = RdataType.WKS
PTR = RdataType.PTR
HINFO = RdataType.HINFO
MINFO = RdataType.MINFO
MX = RdataType.MX
TXT = RdataType.TXT
RP = RdataType.RP
AFSDB = RdataType.AFSDB
X25 = RdataType.X25
ISDN = RdataType.ISDN
RT = RdataType.RT
NSAP = RdataType.NSAP
NSAP_PTR = RdataType.NSAP_PTR
SIG = RdataType.SIG
KEY = RdataType.KEY
PX = RdataType.PX
GPOS = RdataType.GPOS
AAAA = RdataType.AAAA
LOC = RdataType.LOC
NXT = RdataType.NXT
SRV = RdataType.SRV
NAPTR = RdataType.NAPTR
KX = RdataType.KX
CERT = RdataType.CERT
A6 = RdataType.A6
DNAME = RdataType.DNAME
OPT = RdataType.OPT
APL = RdataType.APL
DS = RdataType.DS
SSHFP = RdataType.SSHFP
IPSECKEY = RdataType.IPSECKEY
RRSIG = RdataType.RRSIG
NSEC = RdataType.NSEC
DNSKEY = RdataType.DNSKEY
DHCID = RdataType.DHCID
NSEC3 = RdataType.NSEC3
NSEC3PARAM = RdataType.NSEC3PARAM
TLSA = RdataType.TLSA
SMIMEA = RdataType.SMIMEA
HIP = RdataType.HIP
NINFO = RdataType.NINFO
CDS = RdataType.CDS
CDNSKEY = RdataType.CDNSKEY
OPENPGPKEY = RdataType.OPENPGPKEY
CSYNC = RdataType.CSYNC
ZONEMD = RdataType.ZONEMD
SVCB = RdataType.SVCB
HTTPS = RdataType.HTTPS
DSYNC = RdataType.DSYNC
SPF = RdataType.SPF
UNSPEC = RdataType.UNSPEC
NID = RdataType.NID
L32 = RdataType.L32
L64 = RdataType.L64
LP = RdataType.LP
EUI48 = RdataType.EUI48
EUI64 = RdataType.EUI64
TKEY = RdataType.TKEY
TSIG = RdataType.TSIG
IXFR = RdataType.IXFR
AXFR = RdataType.AXFR
MAILB = RdataType.MAILB
MAILA = RdataType.MAILA
ANY = RdataType.ANY
URI = RdataType.URI
CAA = RdataType.CAA
AVC = RdataType.AVC
AMTRELAY = RdataType.AMTRELAY
RESINFO = RdataType.RESINFO
WALLET = RdataType.WALLET
TA = RdataType.TA
DLV = RdataType.DLV
### END generated RdataType constants