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

venv/Lib/site-packages/dns/rdtypes/nsbase.py

@@ -0,0 +1,63 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-2007, 2009-2011 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.
+
+"""NS-like base classes."""
+
+import dns.exception
+import dns.immutable
+import dns.name
+import dns.rdata
+
+
+@dns.immutable.immutable
+class NSBase(dns.rdata.Rdata):
+    """Base class for rdata that is like an NS record."""
+
+    __slots__ = ["target"]
+
+    def __init__(self, rdclass, rdtype, target):
+        super().__init__(rdclass, rdtype)
+        self.target = self._as_name(target)
+
+    def to_text(self, origin=None, relativize=True, **kw):
+        target = self.target.choose_relativity(origin, relativize)
+        return str(target)
+
+    @classmethod
+    def from_text(
+        cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None
+    ):
+        target = tok.get_name(origin, relativize, relativize_to)
+        return cls(rdclass, rdtype, target)
+
+    def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        self.target.to_wire(file, compress, origin, canonicalize)
+
+    @classmethod
+    def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
+        target = parser.get_name(origin)
+        return cls(rdclass, rdtype, target)
+
+
+@dns.immutable.immutable
+class UncompressedNS(NSBase):
+    """Base class for rdata that is like an NS record, but whose name
+    is not compressed when convert to DNS wire format, and whose
+    digestable form is not downcased."""
+
+    def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        self.target.to_wire(file, None, origin, False)

venv/Lib/site-packages/dns/rdtypes/svcbbase.py

@@ -0,0 +1,587 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import base64
+import enum
+import struct
+from typing import Any, Dict
+
+import dns.enum
+import dns.exception
+import dns.immutable
+import dns.ipv4
+import dns.ipv6
+import dns.name
+import dns.rdata
+import dns.rdtypes.util
+import dns.renderer
+import dns.tokenizer
+import dns.wire
+
+# Until there is an RFC, this module is experimental and may be changed in
+# incompatible ways.
+
+
+class UnknownParamKey(dns.exception.DNSException):
+    """Unknown SVCB ParamKey"""
+
+
+class ParamKey(dns.enum.IntEnum):
+    """SVCB ParamKey"""
+
+    MANDATORY = 0
+    ALPN = 1
+    NO_DEFAULT_ALPN = 2
+    PORT = 3
+    IPV4HINT = 4
+    ECH = 5
+    IPV6HINT = 6
+    DOHPATH = 7
+    OHTTP = 8
+
+    @classmethod
+    def _maximum(cls):
+        return 65535
+
+    @classmethod
+    def _short_name(cls):
+        return "SVCBParamKey"
+
+    @classmethod
+    def _prefix(cls):
+        return "KEY"
+
+    @classmethod
+    def _unknown_exception_class(cls):
+        return UnknownParamKey
+
+
+class Emptiness(enum.IntEnum):
+    NEVER = 0
+    ALWAYS = 1
+    ALLOWED = 2
+
+
+def _validate_key(key):
+    force_generic = False
+    if isinstance(key, bytes):
+        # We decode to latin-1 so we get 0-255 as valid and do NOT interpret
+        # UTF-8 sequences
+        key = key.decode("latin-1")
+    if isinstance(key, str):
+        if key.lower().startswith("key"):
+            force_generic = True
+            if key[3:].startswith("0") and len(key) != 4:
+                # key has leading zeros
+                raise ValueError("leading zeros in key")
+        key = key.replace("-", "_")
+    return (ParamKey.make(key), force_generic)
+
+
+def key_to_text(key):
+    return ParamKey.to_text(key).replace("_", "-").lower()
+
+
+# Like rdata escapify, but escapes ',' too.
+
+_escaped = b'",\\'
+
+
+def _escapify(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 _unescape(value: str) -> bytes:
+    if value == "":
+        return b""
+    unescaped = b""
+    l = len(value)
+    i = 0
+    while i < l:
+        c = value[i]
+        i += 1
+        if c == "\\":
+            if i >= l:  # pragma: no cover   (can't happen via tokenizer get())
+                raise dns.exception.UnexpectedEnd
+            c = value[i]
+            i += 1
+            if c.isdigit():
+                if i >= l:
+                    raise dns.exception.UnexpectedEnd
+                c2 = value[i]
+                i += 1
+                if i >= l:
+                    raise dns.exception.UnexpectedEnd
+                c3 = value[i]
+                i += 1
+                if not (c2.isdigit() and c3.isdigit()):
+                    raise dns.exception.SyntaxError
+                codepoint = int(c) * 100 + int(c2) * 10 + int(c3)
+                if codepoint > 255:
+                    raise dns.exception.SyntaxError
+                unescaped += b"%c" % (codepoint)
+                continue
+        unescaped += c.encode()
+    return unescaped
+
+
+def _split(value):
+    l = len(value)
+    i = 0
+    items = []
+    unescaped = b""
+    while i < l:
+        c = value[i]
+        i += 1
+        if c == ord("\\"):
+            if i >= l:  # pragma: no cover   (can't happen via tokenizer get())
+                raise dns.exception.UnexpectedEnd
+            c = value[i]
+            i += 1
+            unescaped += b"%c" % (c)
+        elif c == ord(","):
+            items.append(unescaped)
+            unescaped = b""
+        else:
+            unescaped += b"%c" % (c)
+    items.append(unescaped)
+    return items
+
+
+@dns.immutable.immutable
+class Param:
+    """Abstract base class for SVCB parameters"""
+
+    @classmethod
+    def emptiness(cls) -> Emptiness:
+        return Emptiness.NEVER
+
+
+@dns.immutable.immutable
+class GenericParam(Param):
+    """Generic SVCB parameter"""
+
+    def __init__(self, value):
+        self.value = dns.rdata.Rdata._as_bytes(value, True)
+
+    @classmethod
+    def emptiness(cls):
+        return Emptiness.ALLOWED
+
+    @classmethod
+    def from_value(cls, value):
+        if value is None or len(value) == 0:
+            return None
+        else:
+            return cls(_unescape(value))
+
+    def to_text(self):
+        return '"' + dns.rdata._escapify(self.value) + '"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        value = parser.get_bytes(parser.remaining())
+        if len(value) == 0:
+            return None
+        else:
+            return cls(value)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        file.write(self.value)
+
+
+@dns.immutable.immutable
+class MandatoryParam(Param):
+    def __init__(self, keys):
+        # check for duplicates
+        keys = sorted([_validate_key(key)[0] for key in keys])
+        prior_k = None
+        for k in keys:
+            if k == prior_k:
+                raise ValueError(f"duplicate key {k:d}")
+            prior_k = k
+            if k == ParamKey.MANDATORY:
+                raise ValueError("listed the mandatory key as mandatory")
+        self.keys = tuple(keys)
+
+    @classmethod
+    def from_value(cls, value):
+        keys = [k.encode() for k in value.split(",")]
+        return cls(keys)
+
+    def to_text(self):
+        return '"' + ",".join([key_to_text(key) for key in self.keys]) + '"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        keys = []
+        last_key = -1
+        while parser.remaining() > 0:
+            key = parser.get_uint16()
+            if key < last_key:
+                raise dns.exception.FormError("manadatory keys not ascending")
+            last_key = key
+            keys.append(key)
+        return cls(keys)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        for key in self.keys:
+            file.write(struct.pack("!H", key))
+
+
+@dns.immutable.immutable
+class ALPNParam(Param):
+    def __init__(self, ids):
+        self.ids = dns.rdata.Rdata._as_tuple(
+            ids, lambda x: dns.rdata.Rdata._as_bytes(x, True, 255, False)
+        )
+
+    @classmethod
+    def from_value(cls, value):
+        return cls(_split(_unescape(value)))
+
+    def to_text(self):
+        value = ",".join([_escapify(id) for id in self.ids])
+        return '"' + dns.rdata._escapify(value.encode()) + '"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        ids = []
+        while parser.remaining() > 0:
+            id = parser.get_counted_bytes()
+            ids.append(id)
+        return cls(ids)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        for id in self.ids:
+            file.write(struct.pack("!B", len(id)))
+            file.write(id)
+
+
+@dns.immutable.immutable
+class NoDefaultALPNParam(Param):
+    # We don't ever expect to instantiate this class, but we need
+    # a from_value() and a from_wire_parser(), so we just return None
+    # from the class methods when things are OK.
+
+    @classmethod
+    def emptiness(cls):
+        return Emptiness.ALWAYS
+
+    @classmethod
+    def from_value(cls, value):
+        if value is None or value == "":
+            return None
+        else:
+            raise ValueError("no-default-alpn with non-empty value")
+
+    def to_text(self):
+        raise NotImplementedError  # pragma: no cover
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        if parser.remaining() != 0:
+            raise dns.exception.FormError
+        return None
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        raise NotImplementedError  # pragma: no cover
+
+
+@dns.immutable.immutable
+class PortParam(Param):
+    def __init__(self, port):
+        self.port = dns.rdata.Rdata._as_uint16(port)
+
+    @classmethod
+    def from_value(cls, value):
+        value = int(value)
+        return cls(value)
+
+    def to_text(self):
+        return f'"{self.port}"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        port = parser.get_uint16()
+        return cls(port)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        file.write(struct.pack("!H", self.port))
+
+
+@dns.immutable.immutable
+class IPv4HintParam(Param):
+    def __init__(self, addresses):
+        self.addresses = dns.rdata.Rdata._as_tuple(
+            addresses, dns.rdata.Rdata._as_ipv4_address
+        )
+
+    @classmethod
+    def from_value(cls, value):
+        addresses = value.split(",")
+        return cls(addresses)
+
+    def to_text(self):
+        return '"' + ",".join(self.addresses) + '"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        addresses = []
+        while parser.remaining() > 0:
+            ip = parser.get_bytes(4)
+            addresses.append(dns.ipv4.inet_ntoa(ip))
+        return cls(addresses)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        for address in self.addresses:
+            file.write(dns.ipv4.inet_aton(address))
+
+
+@dns.immutable.immutable
+class IPv6HintParam(Param):
+    def __init__(self, addresses):
+        self.addresses = dns.rdata.Rdata._as_tuple(
+            addresses, dns.rdata.Rdata._as_ipv6_address
+        )
+
+    @classmethod
+    def from_value(cls, value):
+        addresses = value.split(",")
+        return cls(addresses)
+
+    def to_text(self):
+        return '"' + ",".join(self.addresses) + '"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        addresses = []
+        while parser.remaining() > 0:
+            ip = parser.get_bytes(16)
+            addresses.append(dns.ipv6.inet_ntoa(ip))
+        return cls(addresses)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        for address in self.addresses:
+            file.write(dns.ipv6.inet_aton(address))
+
+
+@dns.immutable.immutable
+class ECHParam(Param):
+    def __init__(self, ech):
+        self.ech = dns.rdata.Rdata._as_bytes(ech, True)
+
+    @classmethod
+    def from_value(cls, value):
+        if "\\" in value:
+            raise ValueError("escape in ECH value")
+        value = base64.b64decode(value.encode())
+        return cls(value)
+
+    def to_text(self):
+        b64 = base64.b64encode(self.ech).decode("ascii")
+        return f'"{b64}"'
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        value = parser.get_bytes(parser.remaining())
+        return cls(value)
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        file.write(self.ech)
+
+
+@dns.immutable.immutable
+class OHTTPParam(Param):
+    # We don't ever expect to instantiate this class, but we need
+    # a from_value() and a from_wire_parser(), so we just return None
+    # from the class methods when things are OK.
+
+    @classmethod
+    def emptiness(cls):
+        return Emptiness.ALWAYS
+
+    @classmethod
+    def from_value(cls, value):
+        if value is None or value == "":
+            return None
+        else:
+            raise ValueError("ohttp with non-empty value")
+
+    def to_text(self):
+        raise NotImplementedError  # pragma: no cover
+
+    @classmethod
+    def from_wire_parser(cls, parser, origin=None):  # pylint: disable=W0613
+        if parser.remaining() != 0:
+            raise dns.exception.FormError
+        return None
+
+    def to_wire(self, file, origin=None):  # pylint: disable=W0613
+        raise NotImplementedError  # pragma: no cover
+
+
+_class_for_key: Dict[ParamKey, Any] = {
+    ParamKey.MANDATORY: MandatoryParam,
+    ParamKey.ALPN: ALPNParam,
+    ParamKey.NO_DEFAULT_ALPN: NoDefaultALPNParam,
+    ParamKey.PORT: PortParam,
+    ParamKey.IPV4HINT: IPv4HintParam,
+    ParamKey.ECH: ECHParam,
+    ParamKey.IPV6HINT: IPv6HintParam,
+    ParamKey.OHTTP: OHTTPParam,
+}
+
+
+def _validate_and_define(params, key, value):
+    (key, force_generic) = _validate_key(_unescape(key))
+    if key in params:
+        raise SyntaxError(f'duplicate key "{key:d}"')
+    cls = _class_for_key.get(key, GenericParam)
+    emptiness = cls.emptiness()
+    if value is None:
+        if emptiness == Emptiness.NEVER:
+            raise SyntaxError("value cannot be empty")
+        value = cls.from_value(value)
+    else:
+        if force_generic:
+            value = cls.from_wire_parser(dns.wire.Parser(_unescape(value)))
+        else:
+            value = cls.from_value(value)
+    params[key] = value
+
+
+@dns.immutable.immutable
+class SVCBBase(dns.rdata.Rdata):
+    """Base class for SVCB-like records"""
+
+    # see: draft-ietf-dnsop-svcb-https-11
+
+    __slots__ = ["priority", "target", "params"]
+
+    def __init__(self, rdclass, rdtype, priority, target, params):
+        super().__init__(rdclass, rdtype)
+        self.priority = self._as_uint16(priority)
+        self.target = self._as_name(target)
+        for k, v in params.items():
+            k = ParamKey.make(k)
+            if not isinstance(v, Param) and v is not None:
+                raise ValueError(f"{k:d} not a Param")
+        self.params = dns.immutable.Dict(params)
+        # Make sure any parameter listed as mandatory is present in the
+        # record.
+        mandatory = params.get(ParamKey.MANDATORY)
+        if mandatory:
+            for key in mandatory.keys:
+                # Note we have to say "not in" as we have None as a value
+                # so a get() and a not None test would be wrong.
+                if key not in params:
+                    raise ValueError(f"key {key:d} declared mandatory but not present")
+        # The no-default-alpn parameter requires the alpn parameter.
+        if ParamKey.NO_DEFAULT_ALPN in params:
+            if ParamKey.ALPN not in params:
+                raise ValueError("no-default-alpn present, but alpn missing")
+
+    def to_text(self, origin=None, relativize=True, **kw):
+        target = self.target.choose_relativity(origin, relativize)
+        params = []
+        for key in sorted(self.params.keys()):
+            value = self.params[key]
+            if value is None:
+                params.append(key_to_text(key))
+            else:
+                kv = key_to_text(key) + "=" + value.to_text()
+                params.append(kv)
+        if len(params) > 0:
+            space = " "
+        else:
+            space = ""
+        return f"{self.priority} {target}{space}{' '.join(params)}"
+
+    @classmethod
+    def from_text(
+        cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None
+    ):
+        priority = tok.get_uint16()
+        target = tok.get_name(origin, relativize, relativize_to)
+        if priority == 0:
+            token = tok.get()
+            if not token.is_eol_or_eof():
+                raise SyntaxError("parameters in AliasMode")
+            tok.unget(token)
+        params = {}
+        while True:
+            token = tok.get()
+            if token.is_eol_or_eof():
+                tok.unget(token)
+                break
+            if token.ttype != dns.tokenizer.IDENTIFIER:
+                raise SyntaxError("parameter is not an identifier")
+            equals = token.value.find("=")
+            if equals == len(token.value) - 1:
+                # 'key=', so next token should be a quoted string without
+                # any intervening whitespace.
+                key = token.value[:-1]
+                token = tok.get(want_leading=True)
+                if token.ttype != dns.tokenizer.QUOTED_STRING:
+                    raise SyntaxError("whitespace after =")
+                value = token.value
+            elif equals > 0:
+                # key=value
+                key = token.value[:equals]
+                value = token.value[equals + 1 :]
+            elif equals == 0:
+                # =key
+                raise SyntaxError('parameter cannot start with "="')
+            else:
+                # key
+                key = token.value
+                value = None
+            _validate_and_define(params, key, value)
+        return cls(rdclass, rdtype, priority, target, params)
+
+    def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        file.write(struct.pack("!H", self.priority))
+        self.target.to_wire(file, None, origin, False)
+        for key in sorted(self.params):
+            file.write(struct.pack("!H", key))
+            value = self.params[key]
+            with dns.renderer.prefixed_length(file, 2):
+                # Note that we're still writing a length of zero if the value is None
+                if value is not None:
+                    value.to_wire(file, origin)
+
+    @classmethod
+    def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
+        priority = parser.get_uint16()
+        target = parser.get_name(origin)
+        if priority == 0 and parser.remaining() != 0:
+            raise dns.exception.FormError("parameters in AliasMode")
+        params = {}
+        prior_key = -1
+        while parser.remaining() > 0:
+            key = parser.get_uint16()
+            if key < prior_key:
+                raise dns.exception.FormError("keys not in order")
+            prior_key = key
+            vlen = parser.get_uint16()
+            pkey = ParamKey.make(key)
+            pcls = _class_for_key.get(pkey, GenericParam)
+            with parser.restrict_to(vlen):
+                value = pcls.from_wire_parser(parser, origin)
+            params[pkey] = value
+        return cls(rdclass, rdtype, priority, target, params)
+
+    def _processing_priority(self):
+        return self.priority
+
+    @classmethod
+    def _processing_order(cls, iterable):
+        return dns.rdtypes.util.priority_processing_order(iterable)

venv/Lib/site-packages/dns/rdtypes/tlsabase.py

@@ -0,0 +1,69 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2005-2007, 2009-2011 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.
+
+import binascii
+import struct
+
+import dns.immutable
+import dns.rdata
+import dns.rdatatype
+
+
+@dns.immutable.immutable
+class TLSABase(dns.rdata.Rdata):
+    """Base class for TLSA and SMIMEA records"""
+
+    # see: RFC 6698
+
+    __slots__ = ["usage", "selector", "mtype", "cert"]
+
+    def __init__(self, rdclass, rdtype, usage, selector, mtype, cert):
+        super().__init__(rdclass, rdtype)
+        self.usage = self._as_uint8(usage)
+        self.selector = self._as_uint8(selector)
+        self.mtype = self._as_uint8(mtype)
+        self.cert = self._as_bytes(cert)
+
+    def to_text(self, origin=None, relativize=True, **kw):
+        kw = kw.copy()
+        chunksize = kw.pop("chunksize", 128)
+        cert = dns.rdata._hexify(
+            self.cert, chunksize=chunksize, **kw  # pyright: ignore
+        )
+        return f"{self.usage} {self.selector} {self.mtype} {cert}"
+
+    @classmethod
+    def from_text(
+        cls, rdclass, rdtype, tok, origin=None, relativize=True, relativize_to=None
+    ):
+        usage = tok.get_uint8()
+        selector = tok.get_uint8()
+        mtype = tok.get_uint8()
+        cert = tok.concatenate_remaining_identifiers().encode()
+        cert = binascii.unhexlify(cert)
+        return cls(rdclass, rdtype, usage, selector, mtype, cert)
+
+    def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        header = struct.pack("!BBB", self.usage, self.selector, self.mtype)
+        file.write(header)
+        file.write(self.cert)
+
+    @classmethod
+    def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
+        header = parser.get_struct("BBB")
+        cert = parser.get_remaining()
+        return cls(rdclass, rdtype, header[0], header[1], header[2], cert)

venv/Lib/site-packages/dns/rdtypes/txtbase.py

@@ -0,0 +1,109 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2006-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.
+
+"""TXT-like base class."""
+
+from typing import Any, Dict, Iterable, Tuple
+
+import dns.exception
+import dns.immutable
+import dns.name
+import dns.rdata
+import dns.rdataclass
+import dns.rdatatype
+import dns.renderer
+import dns.tokenizer
+
+
+@dns.immutable.immutable
+class TXTBase(dns.rdata.Rdata):
+    """Base class for rdata that is like a TXT record (see RFC 1035)."""
+
+    __slots__ = ["strings"]
+
+    def __init__(
+        self,
+        rdclass: dns.rdataclass.RdataClass,
+        rdtype: dns.rdatatype.RdataType,
+        strings: Iterable[bytes | str],
+    ):
+        """Initialize a TXT-like rdata.
+
+        *rdclass*, an ``int`` is the rdataclass of the Rdata.
+
+        *rdtype*, an ``int`` is the rdatatype of the Rdata.
+
+        *strings*, a tuple of ``bytes``
+        """
+        super().__init__(rdclass, rdtype)
+        self.strings: Tuple[bytes] = self._as_tuple(
+            strings, lambda x: self._as_bytes(x, True, 255)
+        )
+        if len(self.strings) == 0:
+            raise ValueError("the list of strings must not be empty")
+
+    def to_text(
+        self,
+        origin: dns.name.Name | None = None,
+        relativize: bool = True,
+        **kw: Dict[str, Any],
+    ) -> str:
+        txt = ""
+        prefix = ""
+        for s in self.strings:
+            txt += f'{prefix}"{dns.rdata._escapify(s)}"'
+            prefix = " "
+        return txt
+
+    @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,
+    ) -> dns.rdata.Rdata:
+        strings = []
+        for token in tok.get_remaining():
+            token = token.unescape_to_bytes()
+            # The 'if' below is always true in the current code, but we
+            # are leaving this check in in case things change some day.
+            if not (
+                token.is_quoted_string() or token.is_identifier()
+            ):  # pragma: no cover
+                raise dns.exception.SyntaxError("expected a string")
+            if len(token.value) > 255:
+                raise dns.exception.SyntaxError("string too long")
+            strings.append(token.value)
+        if len(strings) == 0:
+            raise dns.exception.UnexpectedEnd
+        return cls(rdclass, rdtype, strings)
+
+    def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        for s in self.strings:
+            with dns.renderer.prefixed_length(file, 1):
+                file.write(s)
+
+    @classmethod
+    def from_wire_parser(cls, rdclass, rdtype, parser, origin=None):
+        strings = []
+        while parser.remaining() > 0:
+            s = parser.get_counted_bytes()
+            strings.append(s)
+        return cls(rdclass, rdtype, strings)

venv/Lib/site-packages/dns/rdtypes/util.py

@@ -0,0 +1,269 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2006, 2007, 2009-2011 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.
+
+import collections
+import random
+import struct
+from typing import Any, Iterable, List, Tuple
+
+import dns.exception
+import dns.ipv4
+import dns.ipv6
+import dns.name
+import dns.rdata
+import dns.rdatatype
+import dns.tokenizer
+import dns.wire
+
+
+class Gateway:
+    """A helper class for the IPSECKEY gateway and AMTRELAY relay fields"""
+
+    name = ""
+
+    def __init__(self, type: Any, gateway: str | dns.name.Name | None = None):
+        self.type = dns.rdata.Rdata._as_uint8(type)
+        self.gateway = gateway
+        self._check()
+
+    @classmethod
+    def _invalid_type(cls, gateway_type):
+        return f"invalid {cls.name} type: {gateway_type}"
+
+    def _check(self):
+        if self.type == 0:
+            if self.gateway not in (".", None):
+                raise SyntaxError(f"invalid {self.name} for type 0")
+            self.gateway = None
+        elif self.type == 1:
+            # check that it's OK
+            assert isinstance(self.gateway, str)
+            dns.ipv4.inet_aton(self.gateway)
+        elif self.type == 2:
+            # check that it's OK
+            assert isinstance(self.gateway, str)
+            dns.ipv6.inet_aton(self.gateway)
+        elif self.type == 3:
+            if not isinstance(self.gateway, dns.name.Name):
+                raise SyntaxError(f"invalid {self.name}; not a name")
+        else:
+            raise SyntaxError(self._invalid_type(self.type))
+
+    def to_text(self, origin=None, relativize=True):
+        if self.type == 0:
+            return "."
+        elif self.type in (1, 2):
+            return self.gateway
+        elif self.type == 3:
+            assert isinstance(self.gateway, dns.name.Name)
+            return str(self.gateway.choose_relativity(origin, relativize))
+        else:
+            raise ValueError(self._invalid_type(self.type))  # pragma: no cover
+
+    @classmethod
+    def from_text(
+        cls, gateway_type, tok, origin=None, relativize=True, relativize_to=None
+    ):
+        if gateway_type in (0, 1, 2):
+            gateway = tok.get_string()
+        elif gateway_type == 3:
+            gateway = tok.get_name(origin, relativize, relativize_to)
+        else:
+            raise dns.exception.SyntaxError(
+                cls._invalid_type(gateway_type)
+            )  # pragma: no cover
+        return cls(gateway_type, gateway)
+
+    # pylint: disable=unused-argument
+    def to_wire(self, file, compress=None, origin=None, canonicalize=False):
+        if self.type == 0:
+            pass
+        elif self.type == 1:
+            assert isinstance(self.gateway, str)
+            file.write(dns.ipv4.inet_aton(self.gateway))
+        elif self.type == 2:
+            assert isinstance(self.gateway, str)
+            file.write(dns.ipv6.inet_aton(self.gateway))
+        elif self.type == 3:
+            assert isinstance(self.gateway, dns.name.Name)
+            self.gateway.to_wire(file, None, origin, False)
+        else:
+            raise ValueError(self._invalid_type(self.type))  # pragma: no cover
+
+    # pylint: enable=unused-argument
+
+    @classmethod
+    def from_wire_parser(cls, gateway_type, parser, origin=None):
+        if gateway_type == 0:
+            gateway = None
+        elif gateway_type == 1:
+            gateway = dns.ipv4.inet_ntoa(parser.get_bytes(4))
+        elif gateway_type == 2:
+            gateway = dns.ipv6.inet_ntoa(parser.get_bytes(16))
+        elif gateway_type == 3:
+            gateway = parser.get_name(origin)
+        else:
+            raise dns.exception.FormError(cls._invalid_type(gateway_type))
+        return cls(gateway_type, gateway)
+
+
+class Bitmap:
+    """A helper class for the NSEC/NSEC3/CSYNC type bitmaps"""
+
+    type_name = ""
+
+    def __init__(self, windows: Iterable[Tuple[int, bytes]] | None = None):
+        last_window = -1
+        if windows is None:
+            windows = []
+        self.windows = windows
+        for window, bitmap in self.windows:
+            if not isinstance(window, int):
+                raise ValueError(f"bad {self.type_name} window type")
+            if window <= last_window:
+                raise ValueError(f"bad {self.type_name} window order")
+            if window > 256:
+                raise ValueError(f"bad {self.type_name} window number")
+            last_window = window
+            if not isinstance(bitmap, bytes):
+                raise ValueError(f"bad {self.type_name} octets type")
+            if len(bitmap) == 0 or len(bitmap) > 32:
+                raise ValueError(f"bad {self.type_name} octets")
+
+    def to_text(self) -> str:
+        text = ""
+        for window, bitmap in self.windows:
+            bits = []
+            for i, byte in enumerate(bitmap):
+                for j in range(0, 8):
+                    if byte & (0x80 >> j):
+                        rdtype = dns.rdatatype.RdataType.make(window * 256 + i * 8 + j)
+                        bits.append(dns.rdatatype.to_text(rdtype))
+            text += " " + " ".join(bits)
+        return text
+
+    @classmethod
+    def from_text(cls, tok: "dns.tokenizer.Tokenizer") -> "Bitmap":
+        rdtypes = []
+        for token in tok.get_remaining():
+            rdtype = dns.rdatatype.from_text(token.unescape().value)
+            if rdtype == 0:
+                raise dns.exception.SyntaxError(f"{cls.type_name} with bit 0")
+            rdtypes.append(rdtype)
+        return cls.from_rdtypes(rdtypes)
+
+    @classmethod
+    def from_rdtypes(cls, rdtypes: List[dns.rdatatype.RdataType]) -> "Bitmap":
+        rdtypes = sorted(rdtypes)
+        window = 0
+        octets = 0
+        prior_rdtype = 0
+        bitmap = bytearray(b"\0" * 32)
+        windows = []
+        for rdtype in rdtypes:
+            if rdtype == prior_rdtype:
+                continue
+            prior_rdtype = rdtype
+            new_window = rdtype // 256
+            if new_window != window:
+                if octets != 0:
+                    windows.append((window, bytes(bitmap[0:octets])))
+                bitmap = bytearray(b"\0" * 32)
+                window = new_window
+            offset = rdtype % 256
+            byte = offset // 8
+            bit = offset % 8
+            octets = byte + 1
+            bitmap[byte] = bitmap[byte] | (0x80 >> bit)
+        if octets != 0:
+            windows.append((window, bytes(bitmap[0:octets])))
+        return cls(windows)
+
+    def to_wire(self, file: Any) -> None:
+        for window, bitmap in self.windows:
+            file.write(struct.pack("!BB", window, len(bitmap)))
+            file.write(bitmap)
+
+    @classmethod
+    def from_wire_parser(cls, parser: "dns.wire.Parser") -> "Bitmap":
+        windows = []
+        while parser.remaining() > 0:
+            window = parser.get_uint8()
+            bitmap = parser.get_counted_bytes()
+            windows.append((window, bitmap))
+        return cls(windows)
+
+
+def _priority_table(items):
+    by_priority = collections.defaultdict(list)
+    for rdata in items:
+        by_priority[rdata._processing_priority()].append(rdata)
+    return by_priority
+
+
+def priority_processing_order(iterable):
+    items = list(iterable)
+    if len(items) == 1:
+        return items
+    by_priority = _priority_table(items)
+    ordered = []
+    for k in sorted(by_priority.keys()):
+        rdatas = by_priority[k]
+        random.shuffle(rdatas)
+        ordered.extend(rdatas)
+    return ordered
+
+
+_no_weight = 0.1
+
+
+def weighted_processing_order(iterable):
+    items = list(iterable)
+    if len(items) == 1:
+        return items
+    by_priority = _priority_table(items)
+    ordered = []
+    for k in sorted(by_priority.keys()):
+        rdatas = by_priority[k]
+        total = sum(rdata._processing_weight() or _no_weight for rdata in rdatas)
+        while len(rdatas) > 1:
+            r = random.uniform(0, total)
+            for n, rdata in enumerate(rdatas):  # noqa: B007
+                weight = rdata._processing_weight() or _no_weight
+                if weight > r:
+                    break
+                r -= weight
+            total -= weight  # pyright: ignore[reportPossiblyUnboundVariable]
+            # pylint: disable=undefined-loop-variable
+            ordered.append(rdata)  # pyright: ignore[reportPossiblyUnboundVariable]
+            del rdatas[n]  # pyright: ignore[reportPossiblyUnboundVariable]
+        ordered.append(rdatas[0])
+    return ordered
+
+
+def parse_formatted_hex(formatted, num_chunks, chunk_size, separator):
+    if len(formatted) != num_chunks * (chunk_size + 1) - 1:
+        raise ValueError("invalid formatted hex string")
+    value = b""
+    for _ in range(num_chunks):
+        chunk = formatted[0:chunk_size]
+        value += int(chunk, 16).to_bytes(chunk_size // 2, "big")
+        formatted = formatted[chunk_size:]
+        if len(formatted) > 0 and formatted[0] != separator:
+            raise ValueError("invalid formatted hex string")
+        formatted = formatted[1:]
+    return value