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

venv/Lib/site-packages/dns/set.py

@@ -0,0 +1,308 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-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.
+
+import itertools
+
+
+class Set:
+    """A simple set class.
+
+    This class was originally used to deal with python not having a set class, and
+    originally the class used lists in its implementation.  The ordered and indexable
+    nature of RRsets and Rdatasets is unfortunately widely used in dnspython
+    applications, so for backwards compatibility sets continue to be a custom class, now
+    based on an ordered dictionary.
+    """
+
+    __slots__ = ["items"]
+
+    def __init__(self, items=None):
+        """Initialize the set.
+
+        *items*, an iterable or ``None``, the initial set of items.
+        """
+
+        self.items = dict()
+        if items is not None:
+            for item in items:
+                # This is safe for how we use set, but if other code
+                # subclasses it could be a legitimate issue.
+                self.add(item)  # lgtm[py/init-calls-subclass]
+
+    def __repr__(self):
+        return f"dns.set.Set({repr(list(self.items.keys()))})"  # pragma: no cover
+
+    def add(self, item):
+        """Add an item to the set."""
+
+        if item not in self.items:
+            self.items[item] = None
+
+    def remove(self, item):
+        """Remove an item from the set."""
+
+        try:
+            del self.items[item]
+        except KeyError:
+            raise ValueError
+
+    def discard(self, item):
+        """Remove an item from the set if present."""
+
+        self.items.pop(item, None)
+
+    def pop(self):
+        """Remove an arbitrary item from the set."""
+        (k, _) = self.items.popitem()
+        return k
+
+    def _clone(self) -> "Set":
+        """Make a (shallow) copy of the set.
+
+        There is a 'clone protocol' that subclasses of this class
+        should use.  To make a copy, first call your super's _clone()
+        method, and use the object returned as the new instance.  Then
+        make shallow copies of the attributes defined in the subclass.
+
+        This protocol allows us to write the set algorithms that
+        return new instances (e.g. union) once, and keep using them in
+        subclasses.
+        """
+
+        if hasattr(self, "_clone_class"):
+            cls = self._clone_class  # type: ignore
+        else:
+            cls = self.__class__
+        obj = cls.__new__(cls)
+        obj.items = dict()
+        obj.items.update(self.items)
+        return obj
+
+    def __copy__(self):
+        """Make a (shallow) copy of the set."""
+
+        return self._clone()
+
+    def copy(self):
+        """Make a (shallow) copy of the set."""
+
+        return self._clone()
+
+    def union_update(self, other):
+        """Update the set, adding any elements from other which are not
+        already in the set.
+        """
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        if self is other:  # lgtm[py/comparison-using-is]
+            return
+        for item in other.items:
+            self.add(item)
+
+    def intersection_update(self, other):
+        """Update the set, removing any elements from other which are not
+        in both sets.
+        """
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        if self is other:  # lgtm[py/comparison-using-is]
+            return
+        # we make a copy of the list so that we can remove items from
+        # the list without breaking the iterator.
+        for item in list(self.items):
+            if item not in other.items:
+                del self.items[item]
+
+    def difference_update(self, other):
+        """Update the set, removing any elements from other which are in
+        the set.
+        """
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        if self is other:  # lgtm[py/comparison-using-is]
+            self.items.clear()
+        else:
+            for item in other.items:
+                self.discard(item)
+
+    def symmetric_difference_update(self, other):
+        """Update the set, retaining only elements unique to both sets."""
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        if self is other:  # lgtm[py/comparison-using-is]
+            self.items.clear()
+        else:
+            overlap = self.intersection(other)
+            self.union_update(other)
+            self.difference_update(overlap)
+
+    def union(self, other):
+        """Return a new set which is the union of ``self`` and ``other``.
+
+        Returns the same Set type as this set.
+        """
+
+        obj = self._clone()
+        obj.union_update(other)
+        return obj
+
+    def intersection(self, other):
+        """Return a new set which is the intersection of ``self`` and
+        ``other``.
+
+        Returns the same Set type as this set.
+        """
+
+        obj = self._clone()
+        obj.intersection_update(other)
+        return obj
+
+    def difference(self, other):
+        """Return a new set which ``self`` - ``other``, i.e. the items
+        in ``self`` which are not also in ``other``.
+
+        Returns the same Set type as this set.
+        """
+
+        obj = self._clone()
+        obj.difference_update(other)
+        return obj
+
+    def symmetric_difference(self, other):
+        """Return a new set which (``self`` - ``other``) | (``other``
+        - ``self), ie: the items in either ``self`` or ``other`` which
+        are not contained in their intersection.
+
+        Returns the same Set type as this set.
+        """
+
+        obj = self._clone()
+        obj.symmetric_difference_update(other)
+        return obj
+
+    def __or__(self, other):
+        return self.union(other)
+
+    def __and__(self, other):
+        return self.intersection(other)
+
+    def __add__(self, other):
+        return self.union(other)
+
+    def __sub__(self, other):
+        return self.difference(other)
+
+    def __xor__(self, other):
+        return self.symmetric_difference(other)
+
+    def __ior__(self, other):
+        self.union_update(other)
+        return self
+
+    def __iand__(self, other):
+        self.intersection_update(other)
+        return self
+
+    def __iadd__(self, other):
+        self.union_update(other)
+        return self
+
+    def __isub__(self, other):
+        self.difference_update(other)
+        return self
+
+    def __ixor__(self, other):
+        self.symmetric_difference_update(other)
+        return self
+
+    def update(self, other):
+        """Update the set, adding any elements from other which are not
+        already in the set.
+
+        *other*, the collection of items with which to update the set, which
+        may be any iterable type.
+        """
+
+        for item in other:
+            self.add(item)
+
+    def clear(self):
+        """Make the set empty."""
+        self.items.clear()
+
+    def __eq__(self, other):
+        return self.items == other.items
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __len__(self):
+        return len(self.items)
+
+    def __iter__(self):
+        return iter(self.items)
+
+    def __getitem__(self, i):
+        if isinstance(i, slice):
+            return list(itertools.islice(self.items, i.start, i.stop, i.step))
+        else:
+            return next(itertools.islice(self.items, i, i + 1))
+
+    def __delitem__(self, i):
+        if isinstance(i, slice):
+            for elt in list(self[i]):
+                del self.items[elt]
+        else:
+            del self.items[self[i]]
+
+    def issubset(self, other):
+        """Is this set a subset of *other*?
+
+        Returns a ``bool``.
+        """
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        for item in self.items:
+            if item not in other.items:
+                return False
+        return True
+
+    def issuperset(self, other):
+        """Is this set a superset of *other*?
+
+        Returns a ``bool``.
+        """
+
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        for item in other.items:
+            if item not in self.items:
+                return False
+        return True
+
+    def isdisjoint(self, other):
+        if not isinstance(other, Set):
+            raise ValueError("other must be a Set instance")
+        for item in other.items:
+            if item in self.items:
+                return False
+        return True

venv/Lib/site-packages/dns/tokenizer.py

@@ -0,0 +1,706 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2003-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.
+
+"""Tokenize DNS zone file format"""
+
+import io
+import sys
+from typing import Any, List, Tuple
+
+import dns.exception
+import dns.name
+import dns.ttl
+
+_DELIMITERS = {" ", "\t", "\n", ";", "(", ")", '"'}
+_QUOTING_DELIMITERS = {'"'}
+
+EOF = 0
+EOL = 1
+WHITESPACE = 2
+IDENTIFIER = 3
+QUOTED_STRING = 4
+COMMENT = 5
+DELIMITER = 6
+
+
+class UngetBufferFull(dns.exception.DNSException):
+    """An attempt was made to unget a token when the unget buffer was full."""
+
+
+class Token:
+    """A DNS zone file format token.
+
+    ttype: The token type
+    value: The token value
+    has_escape: Does the token value contain escapes?
+    """
+
+    def __init__(
+        self,
+        ttype: int,
+        value: Any = "",
+        has_escape: bool = False,
+        comment: str | None = None,
+    ):
+        """Initialize a token instance."""
+
+        self.ttype = ttype
+        self.value = value
+        self.has_escape = has_escape
+        self.comment = comment
+
+    def is_eof(self) -> bool:
+        return self.ttype == EOF
+
+    def is_eol(self) -> bool:
+        return self.ttype == EOL
+
+    def is_whitespace(self) -> bool:
+        return self.ttype == WHITESPACE
+
+    def is_identifier(self) -> bool:
+        return self.ttype == IDENTIFIER
+
+    def is_quoted_string(self) -> bool:
+        return self.ttype == QUOTED_STRING
+
+    def is_comment(self) -> bool:
+        return self.ttype == COMMENT
+
+    def is_delimiter(self) -> bool:  # pragma: no cover (we don't return delimiters yet)
+        return self.ttype == DELIMITER
+
+    def is_eol_or_eof(self) -> bool:
+        return self.ttype == EOL or self.ttype == EOF
+
+    def __eq__(self, other):
+        if not isinstance(other, Token):
+            return False
+        return self.ttype == other.ttype and self.value == other.value
+
+    def __ne__(self, other):
+        if not isinstance(other, Token):
+            return True
+        return self.ttype != other.ttype or self.value != other.value
+
+    def __str__(self):
+        return f'{self.ttype} "{self.value}"'
+
+    def unescape(self) -> "Token":
+        if not self.has_escape:
+            return self
+        unescaped = ""
+        l = len(self.value)
+        i = 0
+        while i < l:
+            c = self.value[i]
+            i += 1
+            if c == "\\":
+                if i >= l:  # pragma: no cover   (can't happen via get())
+                    raise dns.exception.UnexpectedEnd
+                c = self.value[i]
+                i += 1
+                if c.isdigit():
+                    if i >= l:
+                        raise dns.exception.UnexpectedEnd
+                    c2 = self.value[i]
+                    i += 1
+                    if i >= l:
+                        raise dns.exception.UnexpectedEnd
+                    c3 = self.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
+                    c = chr(codepoint)
+            unescaped += c
+        return Token(self.ttype, unescaped)
+
+    def unescape_to_bytes(self) -> "Token":
+        # We used to use unescape() for TXT-like records, but this
+        # caused problems as we'd process DNS escapes into Unicode code
+        # points instead of byte values, and then a to_text() of the
+        # processed data would not equal the original input.  For
+        # example, \226 in the TXT record would have a to_text() of
+        # \195\162 because we applied UTF-8 encoding to Unicode code
+        # point 226.
+        #
+        # We now apply escapes while converting directly to bytes,
+        # avoiding this double encoding.
+        #
+        # This code also handles cases where the unicode input has
+        # non-ASCII code-points in it by converting it to UTF-8.  TXT
+        # records aren't defined for Unicode, but this is the best we
+        # can do to preserve meaning.  For example,
+        #
+        #     foo\u200bbar
+        #
+        # (where \u200b is Unicode code point 0x200b) will be treated
+        # as if the input had been the UTF-8 encoding of that string,
+        # namely:
+        #
+        #     foo\226\128\139bar
+        #
+        unescaped = b""
+        l = len(self.value)
+        i = 0
+        while i < l:
+            c = self.value[i]
+            i += 1
+            if c == "\\":
+                if i >= l:  # pragma: no cover   (can't happen via get())
+                    raise dns.exception.UnexpectedEnd
+                c = self.value[i]
+                i += 1
+                if c.isdigit():
+                    if i >= l:
+                        raise dns.exception.UnexpectedEnd
+                    c2 = self.value[i]
+                    i += 1
+                    if i >= l:
+                        raise dns.exception.UnexpectedEnd
+                    c3 = self.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)
+                else:
+                    # Note that as mentioned above, if c is a Unicode
+                    # code point outside of the ASCII range, then this
+                    # += is converting that code point to its UTF-8
+                    # encoding and appending multiple bytes to
+                    # unescaped.
+                    unescaped += c.encode()
+            else:
+                unescaped += c.encode()
+        return Token(self.ttype, bytes(unescaped))
+
+
+class Tokenizer:
+    """A DNS zone file format tokenizer.
+
+    A token object is basically a (type, value) tuple.  The valid
+    types are EOF, EOL, WHITESPACE, IDENTIFIER, QUOTED_STRING,
+    COMMENT, and DELIMITER.
+
+    file: The file to tokenize
+
+    ungotten_char: The most recently ungotten character, or None.
+
+    ungotten_token: The most recently ungotten token, or None.
+
+    multiline: The current multiline level.  This value is increased
+    by one every time a '(' delimiter is read, and decreased by one every time
+    a ')' delimiter is read.
+
+    quoting: This variable is true if the tokenizer is currently
+    reading a quoted string.
+
+    eof: This variable is true if the tokenizer has encountered EOF.
+
+    delimiters: The current delimiter dictionary.
+
+    line_number: The current line number
+
+    filename: A filename that will be returned by the where() method.
+
+    idna_codec: A dns.name.IDNACodec, specifies the IDNA
+    encoder/decoder.  If None, the default IDNA 2003
+    encoder/decoder is used.
+    """
+
+    def __init__(
+        self,
+        f: Any = sys.stdin,
+        filename: str | None = None,
+        idna_codec: dns.name.IDNACodec | None = None,
+    ):
+        """Initialize a tokenizer instance.
+
+        f: The file to tokenize.  The default is sys.stdin.
+        This parameter may also be a string, in which case the tokenizer
+        will take its input from the contents of the string.
+
+        filename: the name of the filename that the where() method
+        will return.
+
+        idna_codec: A dns.name.IDNACodec, specifies the IDNA
+        encoder/decoder.  If None, the default IDNA 2003
+        encoder/decoder is used.
+        """
+
+        if isinstance(f, str):
+            f = io.StringIO(f)
+            if filename is None:
+                filename = "<string>"
+        elif isinstance(f, bytes):
+            f = io.StringIO(f.decode())
+            if filename is None:
+                filename = "<string>"
+        else:
+            if filename is None:
+                if f is sys.stdin:
+                    filename = "<stdin>"
+                else:
+                    filename = "<file>"
+        self.file = f
+        self.ungotten_char: str | None = None
+        self.ungotten_token: Token | None = None
+        self.multiline = 0
+        self.quoting = False
+        self.eof = False
+        self.delimiters = _DELIMITERS
+        self.line_number = 1
+        assert filename is not None
+        self.filename = filename
+        if idna_codec is None:
+            self.idna_codec: dns.name.IDNACodec = dns.name.IDNA_2003
+        else:
+            self.idna_codec = idna_codec
+
+    def _get_char(self) -> str:
+        """Read a character from input."""
+
+        if self.ungotten_char is None:
+            if self.eof:
+                c = ""
+            else:
+                c = self.file.read(1)
+                if c == "":
+                    self.eof = True
+                elif c == "\n":
+                    self.line_number += 1
+        else:
+            c = self.ungotten_char
+            self.ungotten_char = None
+        return c
+
+    def where(self) -> Tuple[str, int]:
+        """Return the current location in the input.
+
+        Returns a (string, int) tuple.  The first item is the filename of
+        the input, the second is the current line number.
+        """
+
+        return (self.filename, self.line_number)
+
+    def _unget_char(self, c: str) -> None:
+        """Unget a character.
+
+        The unget buffer for characters is only one character large; it is
+        an error to try to unget a character when the unget buffer is not
+        empty.
+
+        c: the character to unget
+        raises UngetBufferFull: there is already an ungotten char
+        """
+
+        if self.ungotten_char is not None:
+            # this should never happen!
+            raise UngetBufferFull  # pragma: no cover
+        self.ungotten_char = c
+
+    def skip_whitespace(self) -> int:
+        """Consume input until a non-whitespace character is encountered.
+
+        The non-whitespace character is then ungotten, and the number of
+        whitespace characters consumed is returned.
+
+        If the tokenizer is in multiline mode, then newlines are whitespace.
+
+        Returns the number of characters skipped.
+        """
+
+        skipped = 0
+        while True:
+            c = self._get_char()
+            if c != " " and c != "\t":
+                if (c != "\n") or not self.multiline:
+                    self._unget_char(c)
+                    return skipped
+            skipped += 1
+
+    def get(self, want_leading: bool = False, want_comment: bool = False) -> Token:
+        """Get the next token.
+
+        want_leading: If True, return a WHITESPACE token if the
+        first character read is whitespace.  The default is False.
+
+        want_comment: If True, return a COMMENT token if the
+        first token read is a comment.  The default is False.
+
+        Raises dns.exception.UnexpectedEnd: input ended prematurely
+
+        Raises dns.exception.SyntaxError: input was badly formed
+
+        Returns a Token.
+        """
+
+        if self.ungotten_token is not None:
+            utoken = self.ungotten_token
+            self.ungotten_token = None
+            if utoken.is_whitespace():
+                if want_leading:
+                    return utoken
+            elif utoken.is_comment():
+                if want_comment:
+                    return utoken
+            else:
+                return utoken
+        skipped = self.skip_whitespace()
+        if want_leading and skipped > 0:
+            return Token(WHITESPACE, " ")
+        token = ""
+        ttype = IDENTIFIER
+        has_escape = False
+        while True:
+            c = self._get_char()
+            if c == "" or c in self.delimiters:
+                if c == "" and self.quoting:
+                    raise dns.exception.UnexpectedEnd
+                if token == "" and ttype != QUOTED_STRING:
+                    if c == "(":
+                        self.multiline += 1
+                        self.skip_whitespace()
+                        continue
+                    elif c == ")":
+                        if self.multiline <= 0:
+                            raise dns.exception.SyntaxError
+                        self.multiline -= 1
+                        self.skip_whitespace()
+                        continue
+                    elif c == '"':
+                        if not self.quoting:
+                            self.quoting = True
+                            self.delimiters = _QUOTING_DELIMITERS
+                            ttype = QUOTED_STRING
+                            continue
+                        else:
+                            self.quoting = False
+                            self.delimiters = _DELIMITERS
+                            self.skip_whitespace()
+                            continue
+                    elif c == "\n":
+                        return Token(EOL, "\n")
+                    elif c == ";":
+                        while 1:
+                            c = self._get_char()
+                            if c == "\n" or c == "":
+                                break
+                            token += c
+                        if want_comment:
+                            self._unget_char(c)
+                            return Token(COMMENT, token)
+                        elif c == "":
+                            if self.multiline:
+                                raise dns.exception.SyntaxError(
+                                    "unbalanced parentheses"
+                                )
+                            return Token(EOF, comment=token)
+                        elif self.multiline:
+                            self.skip_whitespace()
+                            token = ""
+                            continue
+                        else:
+                            return Token(EOL, "\n", comment=token)
+                    else:
+                        # This code exists in case we ever want a
+                        # delimiter to be returned.  It never produces
+                        # a token currently.
+                        token = c
+                        ttype = DELIMITER
+                else:
+                    self._unget_char(c)
+                break
+            elif self.quoting and c == "\n":
+                raise dns.exception.SyntaxError("newline in quoted string")
+            elif c == "\\":
+                #
+                # It's an escape.  Put it and the next character into
+                # the token; it will be checked later for goodness.
+                #
+                token += c
+                has_escape = True
+                c = self._get_char()
+                if c == "" or (c == "\n" and not self.quoting):
+                    raise dns.exception.UnexpectedEnd
+            token += c
+        if token == "" and ttype != QUOTED_STRING:
+            if self.multiline:
+                raise dns.exception.SyntaxError("unbalanced parentheses")
+            ttype = EOF
+        return Token(ttype, token, has_escape)
+
+    def unget(self, token: Token) -> None:
+        """Unget a token.
+
+        The unget buffer for tokens is only one token large; it is
+        an error to try to unget a token when the unget buffer is not
+        empty.
+
+        token: the token to unget
+
+        Raises UngetBufferFull: there is already an ungotten token
+        """
+
+        if self.ungotten_token is not None:
+            raise UngetBufferFull
+        self.ungotten_token = token
+
+    def next(self):
+        """Return the next item in an iteration.
+
+        Returns a Token.
+        """
+
+        token = self.get()
+        if token.is_eof():
+            raise StopIteration
+        return token
+
+    __next__ = next
+
+    def __iter__(self):
+        return self
+
+    # Helpers
+
+    def get_int(self, base: int = 10) -> int:
+        """Read the next token and interpret it as an unsigned integer.
+
+        Raises dns.exception.SyntaxError if not an unsigned integer.
+
+        Returns an int.
+        """
+
+        token = self.get().unescape()
+        if not token.is_identifier():
+            raise dns.exception.SyntaxError("expecting an identifier")
+        if not token.value.isdigit():
+            raise dns.exception.SyntaxError("expecting an integer")
+        return int(token.value, base)
+
+    def get_uint8(self) -> int:
+        """Read the next token and interpret it as an 8-bit unsigned
+        integer.
+
+        Raises dns.exception.SyntaxError if not an 8-bit unsigned integer.
+
+        Returns an int.
+        """
+
+        value = self.get_int()
+        if value < 0 or value > 255:
+            raise dns.exception.SyntaxError(f"{value} is not an unsigned 8-bit integer")
+        return value
+
+    def get_uint16(self, base: int = 10) -> int:
+        """Read the next token and interpret it as a 16-bit unsigned
+        integer.
+
+        Raises dns.exception.SyntaxError if not a 16-bit unsigned integer.
+
+        Returns an int.
+        """
+
+        value = self.get_int(base=base)
+        if value < 0 or value > 65535:
+            if base == 8:
+                raise dns.exception.SyntaxError(
+                    f"{value:o} is not an octal unsigned 16-bit integer"
+                )
+            else:
+                raise dns.exception.SyntaxError(
+                    f"{value} is not an unsigned 16-bit integer"
+                )
+        return value
+
+    def get_uint32(self, base: int = 10) -> int:
+        """Read the next token and interpret it as a 32-bit unsigned
+        integer.
+
+        Raises dns.exception.SyntaxError if not a 32-bit unsigned integer.
+
+        Returns an int.
+        """
+
+        value = self.get_int(base=base)
+        if value < 0 or value > 4294967295:
+            raise dns.exception.SyntaxError(
+                f"{value} is not an unsigned 32-bit integer"
+            )
+        return value
+
+    def get_uint48(self, base: int = 10) -> int:
+        """Read the next token and interpret it as a 48-bit unsigned
+        integer.
+
+        Raises dns.exception.SyntaxError if not a 48-bit unsigned integer.
+
+        Returns an int.
+        """
+
+        value = self.get_int(base=base)
+        if value < 0 or value > 281474976710655:
+            raise dns.exception.SyntaxError(
+                f"{value} is not an unsigned 48-bit integer"
+            )
+        return value
+
+    def get_string(self, max_length: int | None = None) -> str:
+        """Read the next token and interpret it as a string.
+
+        Raises dns.exception.SyntaxError if not a string.
+        Raises dns.exception.SyntaxError if token value length
+        exceeds max_length (if specified).
+
+        Returns a string.
+        """
+
+        token = self.get().unescape()
+        if not (token.is_identifier() or token.is_quoted_string()):
+            raise dns.exception.SyntaxError("expecting a string")
+        if max_length and len(token.value) > max_length:
+            raise dns.exception.SyntaxError("string too long")
+        return token.value
+
+    def get_identifier(self) -> str:
+        """Read the next token, which should be an identifier.
+
+        Raises dns.exception.SyntaxError if not an identifier.
+
+        Returns a string.
+        """
+
+        token = self.get().unescape()
+        if not token.is_identifier():
+            raise dns.exception.SyntaxError("expecting an identifier")
+        return token.value
+
+    def get_remaining(self, max_tokens: int | None = None) -> List[Token]:
+        """Return the remaining tokens on the line, until an EOL or EOF is seen.
+
+        max_tokens: If not None, stop after this number of tokens.
+
+        Returns a list of tokens.
+        """
+
+        tokens = []
+        while True:
+            token = self.get()
+            if token.is_eol_or_eof():
+                self.unget(token)
+                break
+            tokens.append(token)
+            if len(tokens) == max_tokens:
+                break
+        return tokens
+
+    def concatenate_remaining_identifiers(self, allow_empty: bool = False) -> str:
+        """Read the remaining tokens on the line, which should be identifiers.
+
+        Raises dns.exception.SyntaxError if there are no remaining tokens,
+        unless `allow_empty=True` is given.
+
+        Raises dns.exception.SyntaxError if a token is seen that is not an
+        identifier.
+
+        Returns a string containing a concatenation of the remaining
+        identifiers.
+        """
+        s = ""
+        while True:
+            token = self.get().unescape()
+            if token.is_eol_or_eof():
+                self.unget(token)
+                break
+            if not token.is_identifier():
+                raise dns.exception.SyntaxError
+            s += token.value
+        if not (allow_empty or s):
+            raise dns.exception.SyntaxError("expecting another identifier")
+        return s
+
+    def as_name(
+        self,
+        token: Token,
+        origin: dns.name.Name | None = None,
+        relativize: bool = False,
+        relativize_to: dns.name.Name | None = None,
+    ) -> dns.name.Name:
+        """Try to interpret the token as a DNS name.
+
+        Raises dns.exception.SyntaxError if not a name.
+
+        Returns a dns.name.Name.
+        """
+        if not token.is_identifier():
+            raise dns.exception.SyntaxError("expecting an identifier")
+        name = dns.name.from_text(token.value, origin, self.idna_codec)
+        return name.choose_relativity(relativize_to or origin, relativize)
+
+    def get_name(
+        self,
+        origin: dns.name.Name | None = None,
+        relativize: bool = False,
+        relativize_to: dns.name.Name | None = None,
+    ) -> dns.name.Name:
+        """Read the next token and interpret it as a DNS name.
+
+        Raises dns.exception.SyntaxError if not a name.
+
+        Returns a dns.name.Name.
+        """
+
+        token = self.get()
+        return self.as_name(token, origin, relativize, relativize_to)
+
+    def get_eol_as_token(self) -> Token:
+        """Read the next token and raise an exception if it isn't EOL or
+        EOF.
+
+        Returns a string.
+        """
+
+        token = self.get()
+        if not token.is_eol_or_eof():
+            raise dns.exception.SyntaxError(
+                f'expected EOL or EOF, got {token.ttype} "{token.value}"'
+            )
+        return token
+
+    def get_eol(self) -> str:
+        return self.get_eol_as_token().value
+
+    def get_ttl(self) -> int:
+        """Read the next token and interpret it as a DNS TTL.
+
+        Raises dns.exception.SyntaxError or dns.ttl.BadTTL if not an
+        identifier or badly formed.
+
+        Returns an int.
+        """
+
+        token = self.get().unescape()
+        if not token.is_identifier():
+            raise dns.exception.SyntaxError("expecting an identifier")
+        return dns.ttl.from_text(token.value)

venv/Lib/site-packages/dns/transaction.py

@@ -0,0 +1,651 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+import collections
+from typing import Any, Callable, Iterator, List, Tuple
+
+import dns.exception
+import dns.name
+import dns.node
+import dns.rdata
+import dns.rdataclass
+import dns.rdataset
+import dns.rdatatype
+import dns.rrset
+import dns.serial
+import dns.ttl
+
+
+class TransactionManager:
+    def reader(self) -> "Transaction":
+        """Begin a read-only transaction."""
+        raise NotImplementedError  # pragma: no cover
+
+    def writer(self, replacement: bool = False) -> "Transaction":
+        """Begin a writable transaction.
+
+        *replacement*, a ``bool``.  If `True`, the content of the
+        transaction completely replaces any prior content.  If False,
+        the default, then the content of the transaction updates the
+        existing content.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def origin_information(
+        self,
+    ) -> Tuple[dns.name.Name | None, bool, dns.name.Name | None]:
+        """Returns a tuple
+
+            (absolute_origin, relativize, effective_origin)
+
+        giving the absolute name of the default origin for any
+        relative domain names, the "effective origin", and whether
+        names should be relativized.  The "effective origin" is the
+        absolute origin if relativize is False, and the empty name if
+        relativize is true.  (The effective origin is provided even
+        though it can be computed from the absolute_origin and
+        relativize setting because it avoids a lot of code
+        duplication.)
+
+        If the returned names are `None`, then no origin information is
+        available.
+
+        This information is used by code working with transactions to
+        allow it to coordinate relativization.  The transaction code
+        itself takes what it gets (i.e. does not change name
+        relativity).
+
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def get_class(self) -> dns.rdataclass.RdataClass:
+        """The class of the transaction manager."""
+        raise NotImplementedError  # pragma: no cover
+
+    def from_wire_origin(self) -> dns.name.Name | None:
+        """Origin to use in from_wire() calls."""
+        (absolute_origin, relativize, _) = self.origin_information()
+        if relativize:
+            return absolute_origin
+        else:
+            return None
+
+
+class DeleteNotExact(dns.exception.DNSException):
+    """Existing data did not match data specified by an exact delete."""
+
+
+class ReadOnly(dns.exception.DNSException):
+    """Tried to write to a read-only transaction."""
+
+
+class AlreadyEnded(dns.exception.DNSException):
+    """Tried to use an already-ended transaction."""
+
+
+def _ensure_immutable_rdataset(rdataset):
+    if rdataset is None or isinstance(rdataset, dns.rdataset.ImmutableRdataset):
+        return rdataset
+    return dns.rdataset.ImmutableRdataset(rdataset)
+
+
+def _ensure_immutable_node(node):
+    if node is None or node.is_immutable():
+        return node
+    return dns.node.ImmutableNode(node)
+
+
+CheckPutRdatasetType = Callable[
+    ["Transaction", dns.name.Name, dns.rdataset.Rdataset], None
+]
+CheckDeleteRdatasetType = Callable[
+    ["Transaction", dns.name.Name, dns.rdatatype.RdataType, dns.rdatatype.RdataType],
+    None,
+]
+CheckDeleteNameType = Callable[["Transaction", dns.name.Name], None]
+
+
+class Transaction:
+    def __init__(
+        self,
+        manager: TransactionManager,
+        replacement: bool = False,
+        read_only: bool = False,
+    ):
+        self.manager = manager
+        self.replacement = replacement
+        self.read_only = read_only
+        self._ended = False
+        self._check_put_rdataset: List[CheckPutRdatasetType] = []
+        self._check_delete_rdataset: List[CheckDeleteRdatasetType] = []
+        self._check_delete_name: List[CheckDeleteNameType] = []
+
+    #
+    # This is the high level API
+    #
+    # Note that we currently use non-immutable types in the return type signature to
+    # avoid covariance problems, e.g. if the caller has a List[Rdataset], mypy will be
+    # unhappy if we return an ImmutableRdataset.
+
+    def get(
+        self,
+        name: dns.name.Name | str | None,
+        rdtype: dns.rdatatype.RdataType | str,
+        covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
+    ) -> dns.rdataset.Rdataset:
+        """Return the rdataset associated with *name*, *rdtype*, and *covers*,
+        or `None` if not found.
+
+        Note that the returned rdataset is immutable.
+        """
+        self._check_ended()
+        if isinstance(name, str):
+            name = dns.name.from_text(name, None)
+        rdtype = dns.rdatatype.RdataType.make(rdtype)
+        covers = dns.rdatatype.RdataType.make(covers)
+        rdataset = self._get_rdataset(name, rdtype, covers)
+        return _ensure_immutable_rdataset(rdataset)
+
+    def get_node(self, name: dns.name.Name) -> dns.node.Node | None:
+        """Return the node at *name*, if any.
+
+        Returns an immutable node or ``None``.
+        """
+        return _ensure_immutable_node(self._get_node(name))
+
+    def _check_read_only(self) -> None:
+        if self.read_only:
+            raise ReadOnly
+
+    def add(self, *args: Any) -> None:
+        """Add records.
+
+        The arguments may be:
+
+            - rrset
+
+            - name, rdataset...
+
+            - name, ttl, rdata...
+        """
+        self._check_ended()
+        self._check_read_only()
+        self._add(False, args)
+
+    def replace(self, *args: Any) -> None:
+        """Replace the existing rdataset at the name with the specified
+        rdataset, or add the specified rdataset if there was no existing
+        rdataset.
+
+        The arguments may be:
+
+            - rrset
+
+            - name, rdataset...
+
+            - name, ttl, rdata...
+
+        Note that if you want to replace the entire node, you should do
+        a delete of the name followed by one or more calls to add() or
+        replace().
+        """
+        self._check_ended()
+        self._check_read_only()
+        self._add(True, args)
+
+    def delete(self, *args: Any) -> None:
+        """Delete records.
+
+        It is not an error if some of the records are not in the existing
+        set.
+
+        The arguments may be:
+
+            - rrset
+
+            - name
+
+            - name, rdatatype, [covers]
+
+            - name, rdataset...
+
+            - name, rdata...
+        """
+        self._check_ended()
+        self._check_read_only()
+        self._delete(False, args)
+
+    def delete_exact(self, *args: Any) -> None:
+        """Delete records.
+
+        The arguments may be:
+
+            - rrset
+
+            - name
+
+            - name, rdatatype, [covers]
+
+            - name, rdataset...
+
+            - name, rdata...
+
+        Raises dns.transaction.DeleteNotExact if some of the records
+        are not in the existing set.
+
+        """
+        self._check_ended()
+        self._check_read_only()
+        self._delete(True, args)
+
+    def name_exists(self, name: dns.name.Name | str) -> bool:
+        """Does the specified name exist?"""
+        self._check_ended()
+        if isinstance(name, str):
+            name = dns.name.from_text(name, None)
+        return self._name_exists(name)
+
+    def update_serial(
+        self,
+        value: int = 1,
+        relative: bool = True,
+        name: dns.name.Name = dns.name.empty,
+    ) -> None:
+        """Update the serial number.
+
+        *value*, an `int`, is an increment if *relative* is `True`, or the
+        actual value to set if *relative* is `False`.
+
+        Raises `KeyError` if there is no SOA rdataset at *name*.
+
+        Raises `ValueError` if *value* is negative or if the increment is
+        so large that it would cause the new serial to be less than the
+        prior value.
+        """
+        self._check_ended()
+        if value < 0:
+            raise ValueError("negative update_serial() value")
+        if isinstance(name, str):
+            name = dns.name.from_text(name, None)
+        rdataset = self._get_rdataset(name, dns.rdatatype.SOA, dns.rdatatype.NONE)
+        if rdataset is None or len(rdataset) == 0:
+            raise KeyError
+        if relative:
+            serial = dns.serial.Serial(rdataset[0].serial) + value
+        else:
+            serial = dns.serial.Serial(value)
+        serial = serial.value  # convert back to int
+        if serial == 0:
+            serial = 1
+        rdata = rdataset[0].replace(serial=serial)
+        new_rdataset = dns.rdataset.from_rdata(rdataset.ttl, rdata)
+        self.replace(name, new_rdataset)
+
+    def __iter__(self):
+        self._check_ended()
+        return self._iterate_rdatasets()
+
+    def changed(self) -> bool:
+        """Has this transaction changed anything?
+
+        For read-only transactions, the result is always `False`.
+
+        For writable transactions, the result is `True` if at some time
+        during the life of the transaction, the content was changed.
+        """
+        self._check_ended()
+        return self._changed()
+
+    def commit(self) -> None:
+        """Commit the transaction.
+
+        Normally transactions are used as context managers and commit
+        or rollback automatically, but it may be done explicitly if needed.
+        A ``dns.transaction.Ended`` exception will be raised if you try
+        to use a transaction after it has been committed or rolled back.
+
+        Raises an exception if the commit fails (in which case the transaction
+        is also rolled back.
+        """
+        self._end(True)
+
+    def rollback(self) -> None:
+        """Rollback the transaction.
+
+        Normally transactions are used as context managers and commit
+        or rollback automatically, but it may be done explicitly if needed.
+        A ``dns.transaction.AlreadyEnded`` exception will be raised if you try
+        to use a transaction after it has been committed or rolled back.
+
+        Rollback cannot otherwise fail.
+        """
+        self._end(False)
+
+    def check_put_rdataset(self, check: CheckPutRdatasetType) -> None:
+        """Call *check* before putting (storing) an rdataset.
+
+        The function is called with the transaction, the name, and the rdataset.
+
+        The check function may safely make non-mutating transaction method
+        calls, but behavior is undefined if mutating transaction methods are
+        called.  The check function should raise an exception if it objects to
+        the put, and otherwise should return ``None``.
+        """
+        self._check_put_rdataset.append(check)
+
+    def check_delete_rdataset(self, check: CheckDeleteRdatasetType) -> None:
+        """Call *check* before deleting an rdataset.
+
+        The function is called with the transaction, the name, the rdatatype,
+        and the covered rdatatype.
+
+        The check function may safely make non-mutating transaction method
+        calls, but behavior is undefined if mutating transaction methods are
+        called.  The check function should raise an exception if it objects to
+        the put, and otherwise should return ``None``.
+        """
+        self._check_delete_rdataset.append(check)
+
+    def check_delete_name(self, check: CheckDeleteNameType) -> None:
+        """Call *check* before putting (storing) an rdataset.
+
+        The function is called with the transaction and the name.
+
+        The check function may safely make non-mutating transaction method
+        calls, but behavior is undefined if mutating transaction methods are
+        called.  The check function should raise an exception if it objects to
+        the put, and otherwise should return ``None``.
+        """
+        self._check_delete_name.append(check)
+
+    def iterate_rdatasets(
+        self,
+    ) -> Iterator[Tuple[dns.name.Name, dns.rdataset.Rdataset]]:
+        """Iterate all the rdatasets in the transaction, returning
+        (`dns.name.Name`, `dns.rdataset.Rdataset`) tuples.
+
+        Note that as is usual with python iterators, adding or removing items
+        while iterating will invalidate the iterator and may raise `RuntimeError`
+        or fail to iterate over all entries."""
+        self._check_ended()
+        return self._iterate_rdatasets()
+
+    def iterate_names(self) -> Iterator[dns.name.Name]:
+        """Iterate all the names in the transaction.
+
+        Note that as is usual with python iterators, adding or removing names
+        while iterating will invalidate the iterator and may raise `RuntimeError`
+        or fail to iterate over all entries."""
+        self._check_ended()
+        return self._iterate_names()
+
+    #
+    # Helper methods
+    #
+
+    def _raise_if_not_empty(self, method, args):
+        if len(args) != 0:
+            raise TypeError(f"extra parameters to {method}")
+
+    def _rdataset_from_args(self, method, deleting, args):
+        try:
+            arg = args.popleft()
+            if isinstance(arg, dns.rrset.RRset):
+                rdataset = arg.to_rdataset()
+            elif isinstance(arg, dns.rdataset.Rdataset):
+                rdataset = arg
+            else:
+                if deleting:
+                    ttl = 0
+                else:
+                    if isinstance(arg, int):
+                        ttl = arg
+                        if ttl > dns.ttl.MAX_TTL:
+                            raise ValueError(f"{method}: TTL value too big")
+                    else:
+                        raise TypeError(f"{method}: expected a TTL")
+                    arg = args.popleft()
+                if isinstance(arg, dns.rdata.Rdata):
+                    rdataset = dns.rdataset.from_rdata(ttl, arg)
+                else:
+                    raise TypeError(f"{method}: expected an Rdata")
+            return rdataset
+        except IndexError:
+            if deleting:
+                return None
+            else:
+                # reraise
+                raise TypeError(f"{method}: expected more arguments")
+
+    def _add(self, replace, args):
+        if replace:
+            method = "replace()"
+        else:
+            method = "add()"
+        try:
+            args = collections.deque(args)
+            arg = args.popleft()
+            if isinstance(arg, str):
+                arg = dns.name.from_text(arg, None)
+            if isinstance(arg, dns.name.Name):
+                name = arg
+                rdataset = self._rdataset_from_args(method, False, args)
+            elif isinstance(arg, dns.rrset.RRset):
+                rrset = arg
+                name = rrset.name
+                # rrsets are also rdatasets, but they don't print the
+                # same and can't be stored in nodes, so convert.
+                rdataset = rrset.to_rdataset()
+            else:
+                raise TypeError(
+                    f"{method} requires a name or RRset as the first argument"
+                )
+            assert rdataset is not None  # for type checkers
+            if rdataset.rdclass != self.manager.get_class():
+                raise ValueError(f"{method} has objects of wrong RdataClass")
+            if rdataset.rdtype == dns.rdatatype.SOA:
+                (_, _, origin) = self._origin_information()
+                if name != origin:
+                    raise ValueError(f"{method} has non-origin SOA")
+            self._raise_if_not_empty(method, args)
+            if not replace:
+                existing = self._get_rdataset(name, rdataset.rdtype, rdataset.covers)
+                if existing is not None:
+                    if isinstance(existing, dns.rdataset.ImmutableRdataset):
+                        trds = dns.rdataset.Rdataset(
+                            existing.rdclass, existing.rdtype, existing.covers
+                        )
+                        trds.update(existing)
+                        existing = trds
+                    rdataset = existing.union(rdataset)
+            self._checked_put_rdataset(name, rdataset)
+        except IndexError:
+            raise TypeError(f"not enough parameters to {method}")
+
+    def _delete(self, exact, args):
+        if exact:
+            method = "delete_exact()"
+        else:
+            method = "delete()"
+        try:
+            args = collections.deque(args)
+            arg = args.popleft()
+            if isinstance(arg, str):
+                arg = dns.name.from_text(arg, None)
+            if isinstance(arg, dns.name.Name):
+                name = arg
+                if len(args) > 0 and (
+                    isinstance(args[0], int) or isinstance(args[0], str)
+                ):
+                    # deleting by type and (optionally) covers
+                    rdtype = dns.rdatatype.RdataType.make(args.popleft())
+                    if len(args) > 0:
+                        covers = dns.rdatatype.RdataType.make(args.popleft())
+                    else:
+                        covers = dns.rdatatype.NONE
+                    self._raise_if_not_empty(method, args)
+                    existing = self._get_rdataset(name, rdtype, covers)
+                    if existing is None:
+                        if exact:
+                            raise DeleteNotExact(f"{method}: missing rdataset")
+                    else:
+                        self._checked_delete_rdataset(name, rdtype, covers)
+                    return
+                else:
+                    rdataset = self._rdataset_from_args(method, True, args)
+            elif isinstance(arg, dns.rrset.RRset):
+                rdataset = arg  # rrsets are also rdatasets
+                name = rdataset.name
+            else:
+                raise TypeError(
+                    f"{method} requires a name or RRset as the first argument"
+                )
+            self._raise_if_not_empty(method, args)
+            if rdataset:
+                if rdataset.rdclass != self.manager.get_class():
+                    raise ValueError(f"{method} has objects of wrong RdataClass")
+                existing = self._get_rdataset(name, rdataset.rdtype, rdataset.covers)
+                if existing is not None:
+                    if exact:
+                        intersection = existing.intersection(rdataset)
+                        if intersection != rdataset:
+                            raise DeleteNotExact(f"{method}: missing rdatas")
+                    rdataset = existing.difference(rdataset)
+                    if len(rdataset) == 0:
+                        self._checked_delete_rdataset(
+                            name, rdataset.rdtype, rdataset.covers
+                        )
+                    else:
+                        self._checked_put_rdataset(name, rdataset)
+                elif exact:
+                    raise DeleteNotExact(f"{method}: missing rdataset")
+            else:
+                if exact and not self._name_exists(name):
+                    raise DeleteNotExact(f"{method}: name not known")
+                self._checked_delete_name(name)
+        except IndexError:
+            raise TypeError(f"not enough parameters to {method}")
+
+    def _check_ended(self):
+        if self._ended:
+            raise AlreadyEnded
+
+    def _end(self, commit):
+        self._check_ended()
+        try:
+            self._end_transaction(commit)
+        finally:
+            self._ended = True
+
+    def _checked_put_rdataset(self, name, rdataset):
+        for check in self._check_put_rdataset:
+            check(self, name, rdataset)
+        self._put_rdataset(name, rdataset)
+
+    def _checked_delete_rdataset(self, name, rdtype, covers):
+        for check in self._check_delete_rdataset:
+            check(self, name, rdtype, covers)
+        self._delete_rdataset(name, rdtype, covers)
+
+    def _checked_delete_name(self, name):
+        for check in self._check_delete_name:
+            check(self, name)
+        self._delete_name(name)
+
+    #
+    # Transactions are context managers.
+    #
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if not self._ended:
+            if exc_type is None:
+                self.commit()
+            else:
+                self.rollback()
+        return False
+
+    #
+    # This is the low level API, which must be implemented by subclasses
+    # of Transaction.
+    #
+
+    def _get_rdataset(self, name, rdtype, covers):
+        """Return the rdataset associated with *name*, *rdtype*, and *covers*,
+        or `None` if not found.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _put_rdataset(self, name, rdataset):
+        """Store the rdataset."""
+        raise NotImplementedError  # pragma: no cover
+
+    def _delete_name(self, name):
+        """Delete all data associated with *name*.
+
+        It is not an error if the name does not exist.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _delete_rdataset(self, name, rdtype, covers):
+        """Delete all data associated with *name*, *rdtype*, and *covers*.
+
+        It is not an error if the rdataset does not exist.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _name_exists(self, name):
+        """Does name exist?
+
+        Returns a bool.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _changed(self):
+        """Has this transaction changed anything?"""
+        raise NotImplementedError  # pragma: no cover
+
+    def _end_transaction(self, commit):
+        """End the transaction.
+
+        *commit*, a bool.  If ``True``, commit the transaction, otherwise
+        roll it back.
+
+        If committing and the commit fails, then roll back and raise an
+        exception.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _set_origin(self, origin):
+        """Set the origin.
+
+        This method is called when reading a possibly relativized
+        source, and an origin setting operation occurs (e.g. $ORIGIN
+        in a zone file).
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    def _iterate_rdatasets(self):
+        """Return an iterator that yields (name, rdataset) tuples."""
+        raise NotImplementedError  # pragma: no cover
+
+    def _iterate_names(self):
+        """Return an iterator that yields a name."""
+        raise NotImplementedError  # pragma: no cover
+
+    def _get_node(self, name):
+        """Return the node at *name*, if any.
+
+        Returns a node or ``None``.
+        """
+        raise NotImplementedError  # pragma: no cover
+
+    #
+    # Low-level API with a default implementation, in case a subclass needs
+    # to override.
+    #
+
+    def _origin_information(self):
+        # This is only used by _add()
+        return self.manager.origin_information()

venv/Lib/site-packages/dns/tsig.py

@@ -0,0 +1,359 @@
+# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
+
+# Copyright (C) 2001-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.
+
+"""DNS TSIG support."""
+
+import base64
+import hashlib
+import hmac
+import struct
+
+import dns.exception
+import dns.name
+import dns.rcode
+import dns.rdataclass
+import dns.rdatatype
+
+
+class BadTime(dns.exception.DNSException):
+    """The current time is not within the TSIG's validity time."""
+
+
+class BadSignature(dns.exception.DNSException):
+    """The TSIG signature fails to verify."""
+
+
+class BadKey(dns.exception.DNSException):
+    """The TSIG record owner name does not match the key."""
+
+
+class BadAlgorithm(dns.exception.DNSException):
+    """The TSIG algorithm does not match the key."""
+
+
+class PeerError(dns.exception.DNSException):
+    """Base class for all TSIG errors generated by the remote peer"""
+
+
+class PeerBadKey(PeerError):
+    """The peer didn't know the key we used"""
+
+
+class PeerBadSignature(PeerError):
+    """The peer didn't like the signature we sent"""
+
+
+class PeerBadTime(PeerError):
+    """The peer didn't like the time we sent"""
+
+
+class PeerBadTruncation(PeerError):
+    """The peer didn't like amount of truncation in the TSIG we sent"""
+
+
+# TSIG Algorithms
+
+HMAC_MD5 = dns.name.from_text("HMAC-MD5.SIG-ALG.REG.INT")
+HMAC_SHA1 = dns.name.from_text("hmac-sha1")
+HMAC_SHA224 = dns.name.from_text("hmac-sha224")
+HMAC_SHA256 = dns.name.from_text("hmac-sha256")
+HMAC_SHA256_128 = dns.name.from_text("hmac-sha256-128")
+HMAC_SHA384 = dns.name.from_text("hmac-sha384")
+HMAC_SHA384_192 = dns.name.from_text("hmac-sha384-192")
+HMAC_SHA512 = dns.name.from_text("hmac-sha512")
+HMAC_SHA512_256 = dns.name.from_text("hmac-sha512-256")
+GSS_TSIG = dns.name.from_text("gss-tsig")
+
+default_algorithm = HMAC_SHA256
+
+mac_sizes = {
+    HMAC_SHA1: 20,
+    HMAC_SHA224: 28,
+    HMAC_SHA256: 32,
+    HMAC_SHA256_128: 16,
+    HMAC_SHA384: 48,
+    HMAC_SHA384_192: 24,
+    HMAC_SHA512: 64,
+    HMAC_SHA512_256: 32,
+    HMAC_MD5: 16,
+    GSS_TSIG: 128,  # This is what we assume to be the worst case!
+}
+
+
+class GSSTSig:
+    """
+    GSS-TSIG TSIG implementation.  This uses the GSS-API context established
+    in the TKEY message handshake to sign messages using GSS-API message
+    integrity codes, per the RFC.
+
+    In order to avoid a direct GSSAPI dependency, the keyring holds a ref
+    to the GSSAPI object required, rather than the key itself.
+    """
+
+    def __init__(self, gssapi_context):
+        self.gssapi_context = gssapi_context
+        self.data = b""
+        self.name = "gss-tsig"
+
+    def update(self, data):
+        self.data += data
+
+    def sign(self):
+        # defer to the GSSAPI function to sign
+        return self.gssapi_context.get_signature(self.data)
+
+    def verify(self, expected):
+        try:
+            # defer to the GSSAPI function to verify
+            return self.gssapi_context.verify_signature(self.data, expected)
+        except Exception:
+            # note the usage of a bare exception
+            raise BadSignature
+
+
+class GSSTSigAdapter:
+    def __init__(self, keyring):
+        self.keyring = keyring
+
+    def __call__(self, message, keyname):
+        if keyname in self.keyring:
+            key = self.keyring[keyname]
+            if isinstance(key, Key) and key.algorithm == GSS_TSIG:
+                if message:
+                    GSSTSigAdapter.parse_tkey_and_step(key, message, keyname)
+            return key
+        else:
+            return None
+
+    @classmethod
+    def parse_tkey_and_step(cls, key, message, keyname):
+        # if the message is a TKEY type, absorb the key material
+        # into the context using step(); this is used to allow the
+        # client to complete the GSSAPI negotiation before attempting
+        # to verify the signed response to a TKEY message exchange
+        try:
+            rrset = message.find_rrset(
+                message.answer, keyname, dns.rdataclass.ANY, dns.rdatatype.TKEY
+            )
+            if rrset:
+                token = rrset[0].key
+                gssapi_context = key.secret
+                return gssapi_context.step(token)
+        except KeyError:
+            pass
+
+
+class HMACTSig:
+    """
+    HMAC TSIG implementation.  This uses the HMAC python module to handle the
+    sign/verify operations.
+    """
+
+    _hashes = {
+        HMAC_SHA1: hashlib.sha1,
+        HMAC_SHA224: hashlib.sha224,
+        HMAC_SHA256: hashlib.sha256,
+        HMAC_SHA256_128: (hashlib.sha256, 128),
+        HMAC_SHA384: hashlib.sha384,
+        HMAC_SHA384_192: (hashlib.sha384, 192),
+        HMAC_SHA512: hashlib.sha512,
+        HMAC_SHA512_256: (hashlib.sha512, 256),
+        HMAC_MD5: hashlib.md5,
+    }
+
+    def __init__(self, key, algorithm):
+        try:
+            hashinfo = self._hashes[algorithm]
+        except KeyError:
+            raise NotImplementedError(f"TSIG algorithm {algorithm} is not supported")
+
+        # create the HMAC context
+        if isinstance(hashinfo, tuple):
+            self.hmac_context = hmac.new(key, digestmod=hashinfo[0])
+            self.size = hashinfo[1]
+        else:
+            self.hmac_context = hmac.new(key, digestmod=hashinfo)
+            self.size = None
+        self.name = self.hmac_context.name
+        if self.size:
+            self.name += f"-{self.size}"
+
+    def update(self, data):
+        return self.hmac_context.update(data)
+
+    def sign(self):
+        # defer to the HMAC digest() function for that digestmod
+        digest = self.hmac_context.digest()
+        if self.size:
+            digest = digest[: (self.size // 8)]
+        return digest
+
+    def verify(self, expected):
+        # re-digest and compare the results
+        mac = self.sign()
+        if not hmac.compare_digest(mac, expected):
+            raise BadSignature
+
+
+def _digest(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=None):
+    """Return a context containing the TSIG rdata for the input parameters
+    @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object
+    @raises ValueError: I{other_data} is too long
+    @raises NotImplementedError: I{algorithm} is not supported
+    """
+
+    first = not (ctx and multi)
+    if first:
+        ctx = get_context(key)
+        if request_mac:
+            ctx.update(struct.pack("!H", len(request_mac)))
+            ctx.update(request_mac)
+    assert ctx is not None  # for type checkers
+    ctx.update(struct.pack("!H", rdata.original_id))
+    ctx.update(wire[2:])
+    if first:
+        ctx.update(key.name.to_digestable())
+        ctx.update(struct.pack("!H", dns.rdataclass.ANY))
+        ctx.update(struct.pack("!I", 0))
+    if time is None:
+        time = rdata.time_signed
+    upper_time = (time >> 32) & 0xFFFF
+    lower_time = time & 0xFFFFFFFF
+    time_encoded = struct.pack("!HIH", upper_time, lower_time, rdata.fudge)
+    other_len = len(rdata.other)
+    if other_len > 65535:
+        raise ValueError("TSIG Other Data is > 65535 bytes")
+    if first:
+        ctx.update(key.algorithm.to_digestable() + time_encoded)
+        ctx.update(struct.pack("!HH", rdata.error, other_len) + rdata.other)
+    else:
+        ctx.update(time_encoded)
+    return ctx
+
+
+def _maybe_start_digest(key, mac, multi):
+    """If this is the first message in a multi-message sequence,
+    start a new context.
+    @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object
+    """
+    if multi:
+        ctx = get_context(key)
+        ctx.update(struct.pack("!H", len(mac)))
+        ctx.update(mac)
+        return ctx
+    else:
+        return None
+
+
+def sign(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=False):
+    """Return a (tsig_rdata, mac, ctx) tuple containing the HMAC TSIG rdata
+    for the input parameters, the HMAC MAC calculated by applying the
+    TSIG signature algorithm, and the TSIG digest context.
+    @rtype: (string, dns.tsig.HMACTSig or dns.tsig.GSSTSig object)
+    @raises ValueError: I{other_data} is too long
+    @raises NotImplementedError: I{algorithm} is not supported
+    """
+
+    ctx = _digest(wire, key, rdata, time, request_mac, ctx, multi)
+    mac = ctx.sign()
+    tsig = rdata.replace(time_signed=time, mac=mac)
+
+    return (tsig, _maybe_start_digest(key, mac, multi))
+
+
+def validate(
+    wire, key, owner, rdata, now, request_mac, tsig_start, ctx=None, multi=False
+):
+    """Validate the specified TSIG rdata against the other input parameters.
+
+    @raises FormError: The TSIG is badly formed.
+    @raises BadTime: There is too much time skew between the client and the
+    server.
+    @raises BadSignature: The TSIG signature did not validate
+    @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object"""
+
+    (adcount,) = struct.unpack("!H", wire[10:12])
+    if adcount == 0:
+        raise dns.exception.FormError
+    adcount -= 1
+    new_wire = wire[0:10] + struct.pack("!H", adcount) + wire[12:tsig_start]
+    if rdata.error != 0:
+        if rdata.error == dns.rcode.BADSIG:
+            raise PeerBadSignature
+        elif rdata.error == dns.rcode.BADKEY:
+            raise PeerBadKey
+        elif rdata.error == dns.rcode.BADTIME:
+            raise PeerBadTime
+        elif rdata.error == dns.rcode.BADTRUNC:
+            raise PeerBadTruncation
+        else:
+            raise PeerError(f"unknown TSIG error code {rdata.error}")
+    if abs(rdata.time_signed - now) > rdata.fudge:
+        raise BadTime
+    if key.name != owner:
+        raise BadKey
+    if key.algorithm != rdata.algorithm:
+        raise BadAlgorithm
+    ctx = _digest(new_wire, key, rdata, None, request_mac, ctx, multi)
+    ctx.verify(rdata.mac)
+    return _maybe_start_digest(key, rdata.mac, multi)
+
+
+def get_context(key):
+    """Returns an HMAC context for the specified key.
+
+    @rtype: HMAC context
+    @raises NotImplementedError: I{algorithm} is not supported
+    """
+
+    if key.algorithm == GSS_TSIG:
+        return GSSTSig(key.secret)
+    else:
+        return HMACTSig(key.secret, key.algorithm)
+
+
+class Key:
+    def __init__(
+        self,
+        name: dns.name.Name | str,
+        secret: bytes | str,
+        algorithm: dns.name.Name | str = default_algorithm,
+    ):
+        if isinstance(name, str):
+            name = dns.name.from_text(name)
+        self.name = name
+        if isinstance(secret, str):
+            secret = base64.decodebytes(secret.encode())
+        self.secret = secret
+        if isinstance(algorithm, str):
+            algorithm = dns.name.from_text(algorithm)
+        self.algorithm = algorithm
+
+    def __eq__(self, other):
+        return (
+            isinstance(other, Key)
+            and self.name == other.name
+            and self.secret == other.secret
+            and self.algorithm == other.algorithm
+        )
+
+    def __repr__(self):
+        r = f"<DNS key name='{self.name}', " + f"algorithm='{self.algorithm}'"
+        if self.algorithm != GSS_TSIG:
+            r += f", secret='{base64.b64encode(self.secret).decode()}'"
+        r += ">"
+        return r

venv/Lib/site-packages/dns/tsigkeyring.py

@@ -0,0 +1,68 @@
+# 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.
+
+"""A place to store TSIG keys."""
+
+import base64
+from typing import Any, Dict
+
+import dns.name
+import dns.tsig
+
+
+def from_text(textring: Dict[str, Any]) -> Dict[dns.name.Name, Any]:
+    """Convert a dictionary containing (textual DNS name, base64 secret)
+    pairs into a binary keyring which has (dns.name.Name, bytes) pairs, or
+    a dictionary containing (textual DNS name, (algorithm, base64 secret))
+    pairs into a binary keyring which has (dns.name.Name, dns.tsig.Key) pairs.
+    @rtype: dict"""
+
+    keyring: Dict[dns.name.Name, Any] = {}
+    for name, value in textring.items():
+        kname = dns.name.from_text(name)
+        if isinstance(value, str):
+            keyring[kname] = dns.tsig.Key(kname, value).secret
+        else:
+            (algorithm, secret) = value
+            keyring[kname] = dns.tsig.Key(kname, secret, algorithm)
+    return keyring
+
+
+def to_text(keyring: Dict[dns.name.Name, Any]) -> Dict[str, Any]:
+    """Convert a dictionary containing (dns.name.Name, dns.tsig.Key) pairs
+    into a text keyring which has (textual DNS name, (textual algorithm,
+    base64 secret)) pairs, or a dictionary containing (dns.name.Name, bytes)
+    pairs into a text keyring which has (textual DNS name, base64 secret) pairs.
+    @rtype: dict"""
+
+    textring = {}
+
+    def b64encode(secret):
+        return base64.encodebytes(secret).decode().rstrip()
+
+    for name, key in keyring.items():
+        tname = name.to_text()
+        if isinstance(key, bytes):
+            textring[tname] = b64encode(key)
+        else:
+            if isinstance(key.secret, bytes):
+                text_secret = b64encode(key.secret)
+            else:
+                text_secret = str(key.secret)
+
+            textring[tname] = (key.algorithm.to_text(), text_secret)
+    return textring