Загрузить файлы в «venv/Lib/site-packages/dns»
This commit is contained in:
98
venv/Lib/site-packages/dns/wire.py
Normal file
98
venv/Lib/site-packages/dns/wire.py
Normal file
@@ -0,0 +1,98 @@
|
||||
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
|
||||
|
||||
import contextlib
|
||||
import struct
|
||||
from typing import Iterator, Optional, Tuple
|
||||
|
||||
import dns.exception
|
||||
import dns.name
|
||||
|
||||
|
||||
class Parser:
|
||||
"""Helper class for parsing DNS wire format."""
|
||||
|
||||
def __init__(self, wire: bytes, current: int = 0):
|
||||
"""Initialize a Parser
|
||||
|
||||
*wire*, a ``bytes`` contains the data to be parsed, and possibly other data.
|
||||
Typically it is the whole message or a slice of it.
|
||||
|
||||
*current*, an `int`, the offset within *wire* where parsing should begin.
|
||||
"""
|
||||
self.wire = wire
|
||||
self.current = 0
|
||||
self.end = len(self.wire)
|
||||
if current:
|
||||
self.seek(current)
|
||||
self.furthest = current
|
||||
|
||||
def remaining(self) -> int:
|
||||
return self.end - self.current
|
||||
|
||||
def get_bytes(self, size: int) -> bytes:
|
||||
assert size >= 0
|
||||
if size > self.remaining():
|
||||
raise dns.exception.FormError
|
||||
output = self.wire[self.current : self.current + size]
|
||||
self.current += size
|
||||
self.furthest = max(self.furthest, self.current)
|
||||
return output
|
||||
|
||||
def get_counted_bytes(self, length_size: int = 1) -> bytes:
|
||||
length = int.from_bytes(self.get_bytes(length_size), "big")
|
||||
return self.get_bytes(length)
|
||||
|
||||
def get_remaining(self) -> bytes:
|
||||
return self.get_bytes(self.remaining())
|
||||
|
||||
def get_uint8(self) -> int:
|
||||
return struct.unpack("!B", self.get_bytes(1))[0]
|
||||
|
||||
def get_uint16(self) -> int:
|
||||
return struct.unpack("!H", self.get_bytes(2))[0]
|
||||
|
||||
def get_uint32(self) -> int:
|
||||
return struct.unpack("!I", self.get_bytes(4))[0]
|
||||
|
||||
def get_uint48(self) -> int:
|
||||
return int.from_bytes(self.get_bytes(6), "big")
|
||||
|
||||
def get_struct(self, format: str) -> Tuple:
|
||||
return struct.unpack(format, self.get_bytes(struct.calcsize(format)))
|
||||
|
||||
def get_name(self, origin: Optional["dns.name.Name"] = None) -> "dns.name.Name":
|
||||
name = dns.name.from_wire_parser(self)
|
||||
if origin:
|
||||
name = name.relativize(origin)
|
||||
return name
|
||||
|
||||
def seek(self, where: int) -> None:
|
||||
# Note that seeking to the end is OK! (If you try to read
|
||||
# after such a seek, you'll get an exception as expected.)
|
||||
if where < 0 or where > self.end:
|
||||
raise dns.exception.FormError
|
||||
self.current = where
|
||||
|
||||
@contextlib.contextmanager
|
||||
def restrict_to(self, size: int) -> Iterator:
|
||||
assert size >= 0
|
||||
if size > self.remaining():
|
||||
raise dns.exception.FormError
|
||||
saved_end = self.end
|
||||
try:
|
||||
self.end = self.current + size
|
||||
yield
|
||||
# We make this check here and not in the finally as we
|
||||
# don't want to raise if we're already raising for some
|
||||
# other reason.
|
||||
if self.current != self.end:
|
||||
raise dns.exception.FormError
|
||||
finally:
|
||||
self.end = saved_end
|
||||
|
||||
@contextlib.contextmanager
|
||||
def restore_furthest(self) -> Iterator:
|
||||
try:
|
||||
yield None
|
||||
finally:
|
||||
self.current = self.furthest
|
||||
356
venv/Lib/site-packages/dns/xfr.py
Normal file
356
venv/Lib/site-packages/dns/xfr.py
Normal file
@@ -0,0 +1,356 @@
|
||||
# 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.
|
||||
|
||||
from typing import Any, List, Tuple, cast
|
||||
|
||||
import dns.edns
|
||||
import dns.exception
|
||||
import dns.message
|
||||
import dns.name
|
||||
import dns.rcode
|
||||
import dns.rdata
|
||||
import dns.rdataset
|
||||
import dns.rdatatype
|
||||
import dns.rdtypes
|
||||
import dns.rdtypes.ANY
|
||||
import dns.rdtypes.ANY.SMIMEA
|
||||
import dns.rdtypes.ANY.SOA
|
||||
import dns.rdtypes.svcbbase
|
||||
import dns.serial
|
||||
import dns.transaction
|
||||
import dns.tsig
|
||||
import dns.zone
|
||||
|
||||
|
||||
class TransferError(dns.exception.DNSException):
|
||||
"""A zone transfer response got a non-zero rcode."""
|
||||
|
||||
def __init__(self, rcode):
|
||||
message = f"Zone transfer error: {dns.rcode.to_text(rcode)}"
|
||||
super().__init__(message)
|
||||
self.rcode = rcode
|
||||
|
||||
|
||||
class SerialWentBackwards(dns.exception.FormError):
|
||||
"""The current serial number is less than the serial we know."""
|
||||
|
||||
|
||||
class UseTCP(dns.exception.DNSException):
|
||||
"""This IXFR cannot be completed with UDP."""
|
||||
|
||||
|
||||
class Inbound:
|
||||
"""
|
||||
State machine for zone transfers.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
txn_manager: dns.transaction.TransactionManager,
|
||||
rdtype: dns.rdatatype.RdataType = dns.rdatatype.AXFR,
|
||||
serial: int | None = None,
|
||||
is_udp: bool = False,
|
||||
):
|
||||
"""Initialize an inbound zone transfer.
|
||||
|
||||
*txn_manager* is a :py:class:`dns.transaction.TransactionManager`.
|
||||
|
||||
*rdtype* can be `dns.rdatatype.AXFR` or `dns.rdatatype.IXFR`
|
||||
|
||||
*serial* is the base serial number for IXFRs, and is required in
|
||||
that case.
|
||||
|
||||
*is_udp*, a ``bool`` indidicates if UDP is being used for this
|
||||
XFR.
|
||||
"""
|
||||
self.txn_manager = txn_manager
|
||||
self.txn: dns.transaction.Transaction | None = None
|
||||
self.rdtype = rdtype
|
||||
if rdtype == dns.rdatatype.IXFR:
|
||||
if serial is None:
|
||||
raise ValueError("a starting serial must be supplied for IXFRs")
|
||||
self.incremental = True
|
||||
elif rdtype == dns.rdatatype.AXFR:
|
||||
if is_udp:
|
||||
raise ValueError("is_udp specified for AXFR")
|
||||
self.incremental = False
|
||||
else:
|
||||
raise ValueError("rdtype is not IXFR or AXFR")
|
||||
self.serial = serial
|
||||
self.is_udp = is_udp
|
||||
(_, _, self.origin) = txn_manager.origin_information()
|
||||
self.soa_rdataset: dns.rdataset.Rdataset | None = None
|
||||
self.done = False
|
||||
self.expecting_SOA = False
|
||||
self.delete_mode = False
|
||||
|
||||
def process_message(self, message: dns.message.Message) -> bool:
|
||||
"""Process one message in the transfer.
|
||||
|
||||
The message should have the same relativization as was specified when
|
||||
the `dns.xfr.Inbound` was created. The message should also have been
|
||||
created with `one_rr_per_rrset=True` because order matters.
|
||||
|
||||
Returns `True` if the transfer is complete, and `False` otherwise.
|
||||
"""
|
||||
if self.txn is None:
|
||||
self.txn = self.txn_manager.writer(not self.incremental)
|
||||
rcode = message.rcode()
|
||||
if rcode != dns.rcode.NOERROR:
|
||||
raise TransferError(rcode)
|
||||
#
|
||||
# We don't require a question section, but if it is present is
|
||||
# should be correct.
|
||||
#
|
||||
if len(message.question) > 0:
|
||||
if message.question[0].name != self.origin:
|
||||
raise dns.exception.FormError("wrong question name")
|
||||
if message.question[0].rdtype != self.rdtype:
|
||||
raise dns.exception.FormError("wrong question rdatatype")
|
||||
answer_index = 0
|
||||
if self.soa_rdataset is None:
|
||||
#
|
||||
# This is the first message. We're expecting an SOA at
|
||||
# the origin.
|
||||
#
|
||||
if not message.answer or message.answer[0].name != self.origin:
|
||||
raise dns.exception.FormError("No answer or RRset not for zone origin")
|
||||
rrset = message.answer[0]
|
||||
rdataset = rrset
|
||||
if rdataset.rdtype != dns.rdatatype.SOA:
|
||||
raise dns.exception.FormError("first RRset is not an SOA")
|
||||
answer_index = 1
|
||||
self.soa_rdataset = rdataset.copy() # pyright: ignore
|
||||
if self.incremental:
|
||||
assert self.soa_rdataset is not None
|
||||
soa = cast(dns.rdtypes.ANY.SOA.SOA, self.soa_rdataset[0])
|
||||
if soa.serial == self.serial:
|
||||
#
|
||||
# We're already up-to-date.
|
||||
#
|
||||
self.done = True
|
||||
elif dns.serial.Serial(soa.serial) < self.serial:
|
||||
# It went backwards!
|
||||
raise SerialWentBackwards
|
||||
else:
|
||||
if self.is_udp and len(message.answer[answer_index:]) == 0:
|
||||
#
|
||||
# There are no more records, so this is the
|
||||
# "truncated" response. Say to use TCP
|
||||
#
|
||||
raise UseTCP
|
||||
#
|
||||
# Note we're expecting another SOA so we can detect
|
||||
# if this IXFR response is an AXFR-style response.
|
||||
#
|
||||
self.expecting_SOA = True
|
||||
#
|
||||
# Process the answer section (other than the initial SOA in
|
||||
# the first message).
|
||||
#
|
||||
for rrset in message.answer[answer_index:]:
|
||||
name = rrset.name
|
||||
rdataset = rrset
|
||||
if self.done:
|
||||
raise dns.exception.FormError("answers after final SOA")
|
||||
assert self.txn is not None # for mypy
|
||||
if rdataset.rdtype == dns.rdatatype.SOA and name == self.origin:
|
||||
#
|
||||
# Every time we see an origin SOA delete_mode inverts
|
||||
#
|
||||
if self.incremental:
|
||||
self.delete_mode = not self.delete_mode
|
||||
#
|
||||
# If this SOA Rdataset is equal to the first we saw
|
||||
# then we're finished. If this is an IXFR we also
|
||||
# check that we're seeing the record in the expected
|
||||
# part of the response.
|
||||
#
|
||||
if rdataset == self.soa_rdataset and (
|
||||
(not self.incremental) or self.delete_mode
|
||||
):
|
||||
#
|
||||
# This is the final SOA
|
||||
#
|
||||
soa = cast(dns.rdtypes.ANY.SOA.SOA, rdataset[0])
|
||||
if self.expecting_SOA:
|
||||
# We got an empty IXFR sequence!
|
||||
raise dns.exception.FormError("empty IXFR sequence")
|
||||
if self.incremental and self.serial != soa.serial:
|
||||
raise dns.exception.FormError("unexpected end of IXFR sequence")
|
||||
self.txn.replace(name, rdataset)
|
||||
self.txn.commit()
|
||||
self.txn = None
|
||||
self.done = True
|
||||
else:
|
||||
#
|
||||
# This is not the final SOA
|
||||
#
|
||||
self.expecting_SOA = False
|
||||
soa = cast(dns.rdtypes.ANY.SOA.SOA, rdataset[0])
|
||||
if self.incremental:
|
||||
if self.delete_mode:
|
||||
# This is the start of an IXFR deletion set
|
||||
if soa.serial != self.serial:
|
||||
raise dns.exception.FormError(
|
||||
"IXFR base serial mismatch"
|
||||
)
|
||||
else:
|
||||
# This is the start of an IXFR addition set
|
||||
self.serial = soa.serial
|
||||
self.txn.replace(name, rdataset)
|
||||
else:
|
||||
# We saw a non-final SOA for the origin in an AXFR.
|
||||
raise dns.exception.FormError("unexpected origin SOA in AXFR")
|
||||
continue
|
||||
if self.expecting_SOA:
|
||||
#
|
||||
# We made an IXFR request and are expecting another
|
||||
# SOA RR, but saw something else, so this must be an
|
||||
# AXFR response.
|
||||
#
|
||||
self.incremental = False
|
||||
self.expecting_SOA = False
|
||||
self.delete_mode = False
|
||||
self.txn.rollback()
|
||||
self.txn = self.txn_manager.writer(True)
|
||||
#
|
||||
# Note we are falling through into the code below
|
||||
# so whatever rdataset this was gets written.
|
||||
#
|
||||
# Add or remove the data
|
||||
if self.delete_mode:
|
||||
self.txn.delete_exact(name, rdataset)
|
||||
else:
|
||||
self.txn.add(name, rdataset)
|
||||
if self.is_udp and not self.done:
|
||||
#
|
||||
# This is a UDP IXFR and we didn't get to done, and we didn't
|
||||
# get the proper "truncated" response
|
||||
#
|
||||
raise dns.exception.FormError("unexpected end of UDP IXFR")
|
||||
return self.done
|
||||
|
||||
#
|
||||
# Inbounds are context managers.
|
||||
#
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if self.txn:
|
||||
self.txn.rollback()
|
||||
return False
|
||||
|
||||
|
||||
def make_query(
|
||||
txn_manager: dns.transaction.TransactionManager,
|
||||
serial: int | None = 0,
|
||||
use_edns: int | bool | None = None,
|
||||
ednsflags: int | None = None,
|
||||
payload: int | None = None,
|
||||
request_payload: int | None = None,
|
||||
options: List[dns.edns.Option] | None = None,
|
||||
keyring: Any = None,
|
||||
keyname: dns.name.Name | None = None,
|
||||
keyalgorithm: dns.name.Name | str = dns.tsig.default_algorithm,
|
||||
) -> Tuple[dns.message.QueryMessage, int | None]:
|
||||
"""Make an AXFR or IXFR query.
|
||||
|
||||
*txn_manager* is a ``dns.transaction.TransactionManager``, typically a
|
||||
``dns.zone.Zone``.
|
||||
|
||||
*serial* is an ``int`` or ``None``. If 0, then IXFR will be
|
||||
attempted using the most recent serial number from the
|
||||
*txn_manager*; it is the caller's responsibility to ensure there
|
||||
are no write transactions active that could invalidate the
|
||||
retrieved serial. If a serial cannot be determined, AXFR will be
|
||||
forced. Other integer values are the starting serial to use.
|
||||
``None`` forces an AXFR.
|
||||
|
||||
Please see the documentation for :py:func:`dns.message.make_query` and
|
||||
:py:func:`dns.message.Message.use_tsig` for details on the other parameters
|
||||
to this function.
|
||||
|
||||
Returns a `(query, serial)` tuple.
|
||||
"""
|
||||
(zone_origin, _, origin) = txn_manager.origin_information()
|
||||
if zone_origin is None:
|
||||
raise ValueError("no zone origin")
|
||||
if serial is None:
|
||||
rdtype = dns.rdatatype.AXFR
|
||||
elif not isinstance(serial, int):
|
||||
raise ValueError("serial is not an integer")
|
||||
elif serial == 0:
|
||||
with txn_manager.reader() as txn:
|
||||
rdataset = txn.get(origin, "SOA")
|
||||
if rdataset:
|
||||
soa = cast(dns.rdtypes.ANY.SOA.SOA, rdataset[0])
|
||||
serial = soa.serial
|
||||
rdtype = dns.rdatatype.IXFR
|
||||
else:
|
||||
serial = None
|
||||
rdtype = dns.rdatatype.AXFR
|
||||
elif serial > 0 and serial < 4294967296:
|
||||
rdtype = dns.rdatatype.IXFR
|
||||
else:
|
||||
raise ValueError("serial out-of-range")
|
||||
rdclass = txn_manager.get_class()
|
||||
q = dns.message.make_query(
|
||||
zone_origin,
|
||||
rdtype,
|
||||
rdclass,
|
||||
use_edns,
|
||||
False,
|
||||
ednsflags,
|
||||
payload,
|
||||
request_payload,
|
||||
options,
|
||||
)
|
||||
if serial is not None:
|
||||
rdata = dns.rdata.from_text(rdclass, "SOA", f". . {serial} 0 0 0 0")
|
||||
rrset = q.find_rrset(
|
||||
q.authority, zone_origin, rdclass, dns.rdatatype.SOA, create=True
|
||||
)
|
||||
rrset.add(rdata, 0)
|
||||
if keyring is not None:
|
||||
q.use_tsig(keyring, keyname, algorithm=keyalgorithm)
|
||||
return (q, serial)
|
||||
|
||||
|
||||
def extract_serial_from_query(query: dns.message.Message) -> int | None:
|
||||
"""Extract the SOA serial number from query if it is an IXFR and return
|
||||
it, otherwise return None.
|
||||
|
||||
*query* is a dns.message.QueryMessage that is an IXFR or AXFR request.
|
||||
|
||||
Raises if the query is not an IXFR or AXFR, or if an IXFR doesn't have
|
||||
an appropriate SOA RRset in the authority section.
|
||||
"""
|
||||
if not isinstance(query, dns.message.QueryMessage):
|
||||
raise ValueError("query not a QueryMessage")
|
||||
question = query.question[0]
|
||||
if question.rdtype == dns.rdatatype.AXFR:
|
||||
return None
|
||||
elif question.rdtype != dns.rdatatype.IXFR:
|
||||
raise ValueError("query is not an AXFR or IXFR")
|
||||
soa_rrset = query.find_rrset(
|
||||
query.authority, question.name, question.rdclass, dns.rdatatype.SOA
|
||||
)
|
||||
soa = cast(dns.rdtypes.ANY.SOA.SOA, soa_rrset[0])
|
||||
return soa.serial
|
||||
1462
venv/Lib/site-packages/dns/zone.py
Normal file
1462
venv/Lib/site-packages/dns/zone.py
Normal file
File diff suppressed because it is too large
Load Diff
756
venv/Lib/site-packages/dns/zonefile.py
Normal file
756
venv/Lib/site-packages/dns/zonefile.py
Normal file
@@ -0,0 +1,756 @@
|
||||
# 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.
|
||||
|
||||
"""DNS Zones."""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from typing import Any, Iterable, List, Set, Tuple, cast
|
||||
|
||||
import dns.exception
|
||||
import dns.grange
|
||||
import dns.name
|
||||
import dns.node
|
||||
import dns.rdata
|
||||
import dns.rdataclass
|
||||
import dns.rdatatype
|
||||
import dns.rdtypes.ANY.SOA
|
||||
import dns.rrset
|
||||
import dns.tokenizer
|
||||
import dns.transaction
|
||||
import dns.ttl
|
||||
|
||||
|
||||
class UnknownOrigin(dns.exception.DNSException):
|
||||
"""Unknown origin"""
|
||||
|
||||
|
||||
class CNAMEAndOtherData(dns.exception.DNSException):
|
||||
"""A node has a CNAME and other data"""
|
||||
|
||||
|
||||
def _check_cname_and_other_data(txn, name, rdataset):
|
||||
rdataset_kind = dns.node.NodeKind.classify_rdataset(rdataset)
|
||||
node = txn.get_node(name)
|
||||
if node is None:
|
||||
# empty nodes are neutral.
|
||||
return
|
||||
node_kind = node.classify()
|
||||
if (
|
||||
node_kind == dns.node.NodeKind.CNAME
|
||||
and rdataset_kind == dns.node.NodeKind.REGULAR
|
||||
):
|
||||
raise CNAMEAndOtherData("rdataset type is not compatible with a CNAME node")
|
||||
elif (
|
||||
node_kind == dns.node.NodeKind.REGULAR
|
||||
and rdataset_kind == dns.node.NodeKind.CNAME
|
||||
):
|
||||
raise CNAMEAndOtherData(
|
||||
"CNAME rdataset is not compatible with a regular data node"
|
||||
)
|
||||
# Otherwise at least one of the node and the rdataset is neutral, so
|
||||
# adding the rdataset is ok
|
||||
|
||||
|
||||
SavedStateType = Tuple[
|
||||
dns.tokenizer.Tokenizer,
|
||||
dns.name.Name | None, # current_origin
|
||||
dns.name.Name | None, # last_name
|
||||
Any | None, # current_file
|
||||
int, # last_ttl
|
||||
bool, # last_ttl_known
|
||||
int, # default_ttl
|
||||
bool,
|
||||
] # default_ttl_known
|
||||
|
||||
|
||||
def _upper_dollarize(s):
|
||||
s = s.upper()
|
||||
if not s.startswith("$"):
|
||||
s = "$" + s
|
||||
return s
|
||||
|
||||
|
||||
class Reader:
|
||||
"""Read a DNS zone file into a transaction."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tok: dns.tokenizer.Tokenizer,
|
||||
rdclass: dns.rdataclass.RdataClass,
|
||||
txn: dns.transaction.Transaction,
|
||||
allow_include: bool = False,
|
||||
allow_directives: bool | Iterable[str] = True,
|
||||
force_name: dns.name.Name | None = None,
|
||||
force_ttl: int | None = None,
|
||||
force_rdclass: dns.rdataclass.RdataClass | None = None,
|
||||
force_rdtype: dns.rdatatype.RdataType | None = None,
|
||||
default_ttl: int | None = None,
|
||||
):
|
||||
self.tok = tok
|
||||
(self.zone_origin, self.relativize, _) = txn.manager.origin_information()
|
||||
self.current_origin = self.zone_origin
|
||||
self.last_ttl = 0
|
||||
self.last_ttl_known = False
|
||||
if force_ttl is not None:
|
||||
default_ttl = force_ttl
|
||||
if default_ttl is None:
|
||||
self.default_ttl = 0
|
||||
self.default_ttl_known = False
|
||||
else:
|
||||
self.default_ttl = default_ttl
|
||||
self.default_ttl_known = True
|
||||
self.last_name = self.current_origin
|
||||
self.zone_rdclass = rdclass
|
||||
self.txn = txn
|
||||
self.saved_state: List[SavedStateType] = []
|
||||
self.current_file: Any | None = None
|
||||
self.allowed_directives: Set[str]
|
||||
if allow_directives is True:
|
||||
self.allowed_directives = {"$GENERATE", "$ORIGIN", "$TTL"}
|
||||
if allow_include:
|
||||
self.allowed_directives.add("$INCLUDE")
|
||||
elif allow_directives is False:
|
||||
# allow_include was ignored in earlier releases if allow_directives was
|
||||
# False, so we continue that.
|
||||
self.allowed_directives = set()
|
||||
else:
|
||||
# Note that if directives are explicitly specified, then allow_include
|
||||
# is ignored.
|
||||
self.allowed_directives = set(_upper_dollarize(d) for d in allow_directives)
|
||||
self.force_name = force_name
|
||||
self.force_ttl = force_ttl
|
||||
self.force_rdclass = force_rdclass
|
||||
self.force_rdtype = force_rdtype
|
||||
self.txn.check_put_rdataset(_check_cname_and_other_data)
|
||||
|
||||
def _eat_line(self):
|
||||
while 1:
|
||||
token = self.tok.get()
|
||||
if token.is_eol_or_eof():
|
||||
break
|
||||
|
||||
def _get_identifier(self):
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
return token
|
||||
|
||||
def _rr_line(self):
|
||||
"""Process one line from a DNS zone file."""
|
||||
token = None
|
||||
# Name
|
||||
if self.force_name is not None:
|
||||
name = self.force_name
|
||||
else:
|
||||
if self.current_origin is None:
|
||||
raise UnknownOrigin
|
||||
token = self.tok.get(want_leading=True)
|
||||
if not token.is_whitespace():
|
||||
self.last_name = self.tok.as_name(token, self.current_origin)
|
||||
else:
|
||||
token = self.tok.get()
|
||||
if token.is_eol_or_eof():
|
||||
# treat leading WS followed by EOL/EOF as if they were EOL/EOF.
|
||||
return
|
||||
self.tok.unget(token)
|
||||
name = self.last_name
|
||||
if name is None:
|
||||
raise dns.exception.SyntaxError("the last used name is undefined")
|
||||
assert self.zone_origin is not None
|
||||
if not name.is_subdomain(self.zone_origin):
|
||||
self._eat_line()
|
||||
return
|
||||
if self.relativize:
|
||||
name = name.relativize(self.zone_origin)
|
||||
|
||||
# TTL
|
||||
if self.force_ttl is not None:
|
||||
ttl = self.force_ttl
|
||||
self.last_ttl = ttl
|
||||
self.last_ttl_known = True
|
||||
else:
|
||||
token = self._get_identifier()
|
||||
ttl = None
|
||||
try:
|
||||
ttl = dns.ttl.from_text(token.value)
|
||||
self.last_ttl = ttl
|
||||
self.last_ttl_known = True
|
||||
token = None
|
||||
except dns.ttl.BadTTL:
|
||||
self.tok.unget(token)
|
||||
|
||||
# Class
|
||||
if self.force_rdclass is not None:
|
||||
rdclass = self.force_rdclass
|
||||
else:
|
||||
token = self._get_identifier()
|
||||
try:
|
||||
rdclass = dns.rdataclass.from_text(token.value)
|
||||
except dns.exception.SyntaxError:
|
||||
raise
|
||||
except Exception:
|
||||
rdclass = self.zone_rdclass
|
||||
self.tok.unget(token)
|
||||
if rdclass != self.zone_rdclass:
|
||||
raise dns.exception.SyntaxError("RR class is not zone's class")
|
||||
|
||||
if ttl is None:
|
||||
# support for <class> <ttl> <type> syntax
|
||||
token = self._get_identifier()
|
||||
ttl = None
|
||||
try:
|
||||
ttl = dns.ttl.from_text(token.value)
|
||||
self.last_ttl = ttl
|
||||
self.last_ttl_known = True
|
||||
token = None
|
||||
except dns.ttl.BadTTL:
|
||||
if self.default_ttl_known:
|
||||
ttl = self.default_ttl
|
||||
elif self.last_ttl_known:
|
||||
ttl = self.last_ttl
|
||||
self.tok.unget(token)
|
||||
|
||||
# Type
|
||||
if self.force_rdtype is not None:
|
||||
rdtype = self.force_rdtype
|
||||
else:
|
||||
token = self._get_identifier()
|
||||
try:
|
||||
rdtype = dns.rdatatype.from_text(token.value)
|
||||
except Exception:
|
||||
raise dns.exception.SyntaxError(f"unknown rdatatype '{token.value}'")
|
||||
|
||||
try:
|
||||
rd = dns.rdata.from_text(
|
||||
rdclass,
|
||||
rdtype,
|
||||
self.tok,
|
||||
self.current_origin,
|
||||
self.relativize,
|
||||
self.zone_origin,
|
||||
)
|
||||
except dns.exception.SyntaxError:
|
||||
# Catch and reraise.
|
||||
raise
|
||||
except Exception:
|
||||
# All exceptions that occur in the processing of rdata
|
||||
# are treated as syntax errors. This is not strictly
|
||||
# correct, but it is correct almost all of the time.
|
||||
# We convert them to syntax errors so that we can emit
|
||||
# helpful filename:line info.
|
||||
(ty, va) = sys.exc_info()[:2]
|
||||
raise dns.exception.SyntaxError(f"caught exception {str(ty)}: {str(va)}")
|
||||
|
||||
if not self.default_ttl_known and rdtype == dns.rdatatype.SOA:
|
||||
# The pre-RFC2308 and pre-BIND9 behavior inherits the zone default
|
||||
# TTL from the SOA minttl if no $TTL statement is present before the
|
||||
# SOA is parsed.
|
||||
soa_rd = cast(dns.rdtypes.ANY.SOA.SOA, rd)
|
||||
self.default_ttl = soa_rd.minimum
|
||||
self.default_ttl_known = True
|
||||
if ttl is None:
|
||||
# if we didn't have a TTL on the SOA, set it!
|
||||
ttl = soa_rd.minimum
|
||||
|
||||
# TTL check. We had to wait until now to do this as the SOA RR's
|
||||
# own TTL can be inferred from its minimum.
|
||||
if ttl is None:
|
||||
raise dns.exception.SyntaxError("Missing default TTL value")
|
||||
|
||||
self.txn.add(name, ttl, rd)
|
||||
|
||||
def _parse_modify(self, side: str) -> Tuple[str, str, int, int, str]:
|
||||
# Here we catch everything in '{' '}' in a group so we can replace it
|
||||
# with ''.
|
||||
is_generate1 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+),(.)}).*$")
|
||||
is_generate2 = re.compile(r"^.*\$({(\+|-?)(\d+)}).*$")
|
||||
is_generate3 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+)}).*$")
|
||||
# Sometimes there are modifiers in the hostname. These come after
|
||||
# the dollar sign. They are in the form: ${offset[,width[,base]]}.
|
||||
# Make names
|
||||
mod = ""
|
||||
sign = "+"
|
||||
offset = "0"
|
||||
width = "0"
|
||||
base = "d"
|
||||
g1 = is_generate1.match(side)
|
||||
if g1:
|
||||
mod, sign, offset, width, base = g1.groups()
|
||||
if sign == "":
|
||||
sign = "+"
|
||||
else:
|
||||
g2 = is_generate2.match(side)
|
||||
if g2:
|
||||
mod, sign, offset = g2.groups()
|
||||
if sign == "":
|
||||
sign = "+"
|
||||
width = "0"
|
||||
base = "d"
|
||||
else:
|
||||
g3 = is_generate3.match(side)
|
||||
if g3:
|
||||
mod, sign, offset, width = g3.groups()
|
||||
if sign == "":
|
||||
sign = "+"
|
||||
base = "d"
|
||||
|
||||
ioffset = int(offset)
|
||||
iwidth = int(width)
|
||||
|
||||
if sign not in ["+", "-"]:
|
||||
raise dns.exception.SyntaxError(f"invalid offset sign {sign}")
|
||||
if base not in ["d", "o", "x", "X", "n", "N"]:
|
||||
raise dns.exception.SyntaxError(f"invalid type {base}")
|
||||
|
||||
return mod, sign, ioffset, iwidth, base
|
||||
|
||||
def _generate_line(self):
|
||||
# range lhs [ttl] [class] type rhs [ comment ]
|
||||
"""Process one line containing the GENERATE statement from a DNS
|
||||
zone file."""
|
||||
if self.current_origin is None:
|
||||
raise UnknownOrigin
|
||||
|
||||
token = self.tok.get()
|
||||
# Range (required)
|
||||
try:
|
||||
start, stop, step = dns.grange.from_text(token.value)
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
except Exception:
|
||||
raise dns.exception.SyntaxError
|
||||
|
||||
# lhs (required)
|
||||
try:
|
||||
lhs = token.value
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
except Exception:
|
||||
raise dns.exception.SyntaxError
|
||||
|
||||
# TTL
|
||||
try:
|
||||
ttl = dns.ttl.from_text(token.value)
|
||||
self.last_ttl = ttl
|
||||
self.last_ttl_known = True
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
except dns.ttl.BadTTL:
|
||||
if not (self.last_ttl_known or self.default_ttl_known):
|
||||
raise dns.exception.SyntaxError("Missing default TTL value")
|
||||
if self.default_ttl_known:
|
||||
ttl = self.default_ttl
|
||||
elif self.last_ttl_known:
|
||||
ttl = self.last_ttl
|
||||
else:
|
||||
# We don't go to the extra "look at the SOA" level of effort for
|
||||
# $GENERATE, because the user really ought to have defined a TTL
|
||||
# somehow!
|
||||
raise dns.exception.SyntaxError("Missing default TTL value")
|
||||
|
||||
# Class
|
||||
try:
|
||||
rdclass = dns.rdataclass.from_text(token.value)
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
except dns.exception.SyntaxError:
|
||||
raise dns.exception.SyntaxError
|
||||
except Exception:
|
||||
rdclass = self.zone_rdclass
|
||||
if rdclass != self.zone_rdclass:
|
||||
raise dns.exception.SyntaxError("RR class is not zone's class")
|
||||
# Type
|
||||
try:
|
||||
rdtype = dns.rdatatype.from_text(token.value)
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError
|
||||
except Exception:
|
||||
raise dns.exception.SyntaxError(f"unknown rdatatype '{token.value}'")
|
||||
|
||||
# rhs (required)
|
||||
rhs = token.value
|
||||
|
||||
def _calculate_index(counter: int, offset_sign: str, offset: int) -> int:
|
||||
"""Calculate the index from the counter and offset."""
|
||||
if offset_sign == "-":
|
||||
offset *= -1
|
||||
return counter + offset
|
||||
|
||||
def _format_index(index: int, base: str, width: int) -> str:
|
||||
"""Format the index with the given base, and zero-fill it
|
||||
to the given width."""
|
||||
if base in ["d", "o", "x", "X"]:
|
||||
return format(index, base).zfill(width)
|
||||
|
||||
# base can only be n or N here
|
||||
hexa = _format_index(index, "x", width)
|
||||
nibbles = ".".join(hexa[::-1])[:width]
|
||||
if base == "N":
|
||||
nibbles = nibbles.upper()
|
||||
return nibbles
|
||||
|
||||
lmod, lsign, loffset, lwidth, lbase = self._parse_modify(lhs)
|
||||
rmod, rsign, roffset, rwidth, rbase = self._parse_modify(rhs)
|
||||
for i in range(start, stop + 1, step):
|
||||
# +1 because bind is inclusive and python is exclusive
|
||||
|
||||
lindex = _calculate_index(i, lsign, loffset)
|
||||
rindex = _calculate_index(i, rsign, roffset)
|
||||
|
||||
lzfindex = _format_index(lindex, lbase, lwidth)
|
||||
rzfindex = _format_index(rindex, rbase, rwidth)
|
||||
|
||||
name = lhs.replace(f"${lmod}", lzfindex)
|
||||
rdata = rhs.replace(f"${rmod}", rzfindex)
|
||||
|
||||
self.last_name = dns.name.from_text(
|
||||
name, self.current_origin, self.tok.idna_codec
|
||||
)
|
||||
name = self.last_name
|
||||
assert self.zone_origin is not None
|
||||
if not name.is_subdomain(self.zone_origin):
|
||||
self._eat_line()
|
||||
return
|
||||
if self.relativize:
|
||||
name = name.relativize(self.zone_origin)
|
||||
|
||||
try:
|
||||
rd = dns.rdata.from_text(
|
||||
rdclass,
|
||||
rdtype,
|
||||
rdata,
|
||||
self.current_origin,
|
||||
self.relativize,
|
||||
self.zone_origin,
|
||||
)
|
||||
except dns.exception.SyntaxError:
|
||||
# Catch and reraise.
|
||||
raise
|
||||
except Exception:
|
||||
# All exceptions that occur in the processing of rdata
|
||||
# are treated as syntax errors. This is not strictly
|
||||
# correct, but it is correct almost all of the time.
|
||||
# We convert them to syntax errors so that we can emit
|
||||
# helpful filename:line info.
|
||||
(ty, va) = sys.exc_info()[:2]
|
||||
raise dns.exception.SyntaxError(
|
||||
f"caught exception {str(ty)}: {str(va)}"
|
||||
)
|
||||
|
||||
self.txn.add(name, ttl, rd)
|
||||
|
||||
def read(self) -> None:
|
||||
"""Read a DNS zone file and build a zone object.
|
||||
|
||||
@raises dns.zone.NoSOA: No SOA RR was found at the zone origin
|
||||
@raises dns.zone.NoNS: No NS RRset was found at the zone origin
|
||||
"""
|
||||
|
||||
try:
|
||||
while 1:
|
||||
token = self.tok.get(True, True)
|
||||
if token.is_eof():
|
||||
if self.current_file is not None:
|
||||
self.current_file.close()
|
||||
if len(self.saved_state) > 0:
|
||||
(
|
||||
self.tok,
|
||||
self.current_origin,
|
||||
self.last_name,
|
||||
self.current_file,
|
||||
self.last_ttl,
|
||||
self.last_ttl_known,
|
||||
self.default_ttl,
|
||||
self.default_ttl_known,
|
||||
) = self.saved_state.pop(-1)
|
||||
continue
|
||||
break
|
||||
elif token.is_eol():
|
||||
continue
|
||||
elif token.is_comment():
|
||||
self.tok.get_eol()
|
||||
continue
|
||||
elif token.value[0] == "$" and len(self.allowed_directives) > 0:
|
||||
# Note that we only run directive processing code if at least
|
||||
# one directive is allowed in order to be backwards compatible
|
||||
c = token.value.upper()
|
||||
if c not in self.allowed_directives:
|
||||
raise dns.exception.SyntaxError(
|
||||
f"zone file directive '{c}' is not allowed"
|
||||
)
|
||||
if c == "$TTL":
|
||||
token = self.tok.get()
|
||||
if not token.is_identifier():
|
||||
raise dns.exception.SyntaxError("bad $TTL")
|
||||
self.default_ttl = dns.ttl.from_text(token.value)
|
||||
self.default_ttl_known = True
|
||||
self.tok.get_eol()
|
||||
elif c == "$ORIGIN":
|
||||
self.current_origin = self.tok.get_name()
|
||||
self.tok.get_eol()
|
||||
if self.zone_origin is None:
|
||||
self.zone_origin = self.current_origin
|
||||
self.txn._set_origin(self.current_origin)
|
||||
elif c == "$INCLUDE":
|
||||
token = self.tok.get()
|
||||
filename = token.value
|
||||
token = self.tok.get()
|
||||
new_origin: dns.name.Name | None
|
||||
if token.is_identifier():
|
||||
new_origin = dns.name.from_text(
|
||||
token.value, self.current_origin, self.tok.idna_codec
|
||||
)
|
||||
self.tok.get_eol()
|
||||
elif not token.is_eol_or_eof():
|
||||
raise dns.exception.SyntaxError("bad origin in $INCLUDE")
|
||||
else:
|
||||
new_origin = self.current_origin
|
||||
self.saved_state.append(
|
||||
(
|
||||
self.tok,
|
||||
self.current_origin,
|
||||
self.last_name,
|
||||
self.current_file,
|
||||
self.last_ttl,
|
||||
self.last_ttl_known,
|
||||
self.default_ttl,
|
||||
self.default_ttl_known,
|
||||
)
|
||||
)
|
||||
self.current_file = open(filename, encoding="utf-8")
|
||||
self.tok = dns.tokenizer.Tokenizer(self.current_file, filename)
|
||||
self.current_origin = new_origin
|
||||
elif c == "$GENERATE":
|
||||
self._generate_line()
|
||||
else:
|
||||
raise dns.exception.SyntaxError(
|
||||
f"Unknown zone file directive '{c}'"
|
||||
)
|
||||
continue
|
||||
self.tok.unget(token)
|
||||
self._rr_line()
|
||||
except dns.exception.SyntaxError as detail:
|
||||
(filename, line_number) = self.tok.where()
|
||||
if detail is None:
|
||||
detail = "syntax error"
|
||||
ex = dns.exception.SyntaxError(f"{filename}:{line_number}: {detail}")
|
||||
tb = sys.exc_info()[2]
|
||||
raise ex.with_traceback(tb) from None
|
||||
|
||||
|
||||
class RRsetsReaderTransaction(dns.transaction.Transaction):
|
||||
def __init__(self, manager, replacement, read_only):
|
||||
assert not read_only
|
||||
super().__init__(manager, replacement, read_only)
|
||||
self.rdatasets = {}
|
||||
|
||||
def _get_rdataset(self, name, rdtype, covers):
|
||||
return self.rdatasets.get((name, rdtype, covers))
|
||||
|
||||
def _get_node(self, name):
|
||||
rdatasets = []
|
||||
for (rdataset_name, _, _), rdataset in self.rdatasets.items():
|
||||
if name == rdataset_name:
|
||||
rdatasets.append(rdataset)
|
||||
if len(rdatasets) == 0:
|
||||
return None
|
||||
node = dns.node.Node()
|
||||
node.rdatasets = rdatasets
|
||||
return node
|
||||
|
||||
def _put_rdataset(self, name, rdataset):
|
||||
self.rdatasets[(name, rdataset.rdtype, rdataset.covers)] = rdataset
|
||||
|
||||
def _delete_name(self, name):
|
||||
# First remove any changes involving the name
|
||||
remove = []
|
||||
for key in self.rdatasets:
|
||||
if key[0] == name:
|
||||
remove.append(key)
|
||||
if len(remove) > 0:
|
||||
for key in remove:
|
||||
del self.rdatasets[key]
|
||||
|
||||
def _delete_rdataset(self, name, rdtype, covers):
|
||||
try:
|
||||
del self.rdatasets[(name, rdtype, covers)]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
def _name_exists(self, name):
|
||||
for n, _, _ in self.rdatasets:
|
||||
if n == name:
|
||||
return True
|
||||
return False
|
||||
|
||||
def _changed(self):
|
||||
return len(self.rdatasets) > 0
|
||||
|
||||
def _end_transaction(self, commit):
|
||||
if commit and self._changed():
|
||||
rrsets = []
|
||||
for (name, _, _), rdataset in self.rdatasets.items():
|
||||
rrset = dns.rrset.RRset(
|
||||
name, rdataset.rdclass, rdataset.rdtype, rdataset.covers
|
||||
)
|
||||
rrset.update(rdataset)
|
||||
rrsets.append(rrset)
|
||||
self.manager.set_rrsets(rrsets) # pyright: ignore
|
||||
|
||||
def _set_origin(self, origin):
|
||||
pass
|
||||
|
||||
def _iterate_rdatasets(self):
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
def _iterate_names(self):
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
|
||||
class RRSetsReaderManager(dns.transaction.TransactionManager):
|
||||
def __init__(
|
||||
self,
|
||||
origin: dns.name.Name | None = dns.name.root,
|
||||
relativize: bool = False,
|
||||
rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
|
||||
):
|
||||
self.origin = origin
|
||||
self.relativize = relativize
|
||||
self.rdclass = rdclass
|
||||
self.rrsets: List[dns.rrset.RRset] = []
|
||||
|
||||
def reader(self): # pragma: no cover
|
||||
raise NotImplementedError
|
||||
|
||||
def writer(self, replacement=False):
|
||||
assert replacement is True
|
||||
return RRsetsReaderTransaction(self, True, False)
|
||||
|
||||
def get_class(self):
|
||||
return self.rdclass
|
||||
|
||||
def origin_information(self):
|
||||
if self.relativize:
|
||||
effective = dns.name.empty
|
||||
else:
|
||||
effective = self.origin
|
||||
return (self.origin, self.relativize, effective)
|
||||
|
||||
def set_rrsets(self, rrsets: List[dns.rrset.RRset]) -> None:
|
||||
self.rrsets = rrsets
|
||||
|
||||
|
||||
def read_rrsets(
|
||||
text: Any,
|
||||
name: dns.name.Name | str | None = None,
|
||||
ttl: int | None = None,
|
||||
rdclass: dns.rdataclass.RdataClass | str | None = dns.rdataclass.IN,
|
||||
default_rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
|
||||
rdtype: dns.rdatatype.RdataType | str | None = None,
|
||||
default_ttl: int | str | None = None,
|
||||
idna_codec: dns.name.IDNACodec | None = None,
|
||||
origin: dns.name.Name | str | None = dns.name.root,
|
||||
relativize: bool = False,
|
||||
) -> List[dns.rrset.RRset]:
|
||||
"""Read one or more rrsets from the specified text, possibly subject
|
||||
to restrictions.
|
||||
|
||||
*text*, a file object or a string, is the input to process.
|
||||
|
||||
*name*, a string, ``dns.name.Name``, or ``None``, is the owner name of
|
||||
the rrset. If not ``None``, then the owner name is "forced", and the
|
||||
input must not specify an owner name. If ``None``, then any owner names
|
||||
are allowed and must be present in the input.
|
||||
|
||||
*ttl*, an ``int``, string, or None. If not ``None``, the the TTL is
|
||||
forced to be the specified value and the input must not specify a TTL.
|
||||
If ``None``, then a TTL may be specified in the input. If it is not
|
||||
specified, then the *default_ttl* will be used.
|
||||
|
||||
*rdclass*, a ``dns.rdataclass.RdataClass``, string, or ``None``. If
|
||||
not ``None``, then the class is forced to the specified value, and the
|
||||
input must not specify a class. If ``None``, then the input may specify
|
||||
a class that matches *default_rdclass*. Note that it is not possible to
|
||||
return rrsets with differing classes; specifying ``None`` for the class
|
||||
simply allows the user to optionally type a class as that may be convenient
|
||||
when cutting and pasting.
|
||||
|
||||
*default_rdclass*, a ``dns.rdataclass.RdataClass`` or string. The class
|
||||
of the returned rrsets.
|
||||
|
||||
*rdtype*, a ``dns.rdatatype.RdataType``, string, or ``None``. If not
|
||||
``None``, then the type is forced to the specified value, and the
|
||||
input must not specify a type. If ``None``, then a type must be present
|
||||
for each RR.
|
||||
|
||||
*default_ttl*, an ``int``, string, or ``None``. If not ``None``, then if
|
||||
the TTL is not forced and is not specified, then this value will be used.
|
||||
if ``None``, then if the TTL is not forced an error will occur if the TTL
|
||||
is not specified.
|
||||
|
||||
*idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA
|
||||
encoder/decoder. If ``None``, the default IDNA 2003 encoder/decoder
|
||||
is used. Note that codecs only apply to the owner name; dnspython does
|
||||
not do IDNA for names in rdata, as there is no IDNA zonefile format.
|
||||
|
||||
*origin*, a string, ``dns.name.Name``, or ``None``, is the origin for any
|
||||
relative names in the input, and also the origin to relativize to if
|
||||
*relativize* is ``True``.
|
||||
|
||||
*relativize*, a bool. If ``True``, names are relativized to the *origin*;
|
||||
if ``False`` then any relative names in the input are made absolute by
|
||||
appending the *origin*.
|
||||
"""
|
||||
if isinstance(origin, str):
|
||||
origin = dns.name.from_text(origin, dns.name.root, idna_codec)
|
||||
if isinstance(name, str):
|
||||
name = dns.name.from_text(name, origin, idna_codec)
|
||||
if isinstance(ttl, str):
|
||||
ttl = dns.ttl.from_text(ttl)
|
||||
if isinstance(default_ttl, str):
|
||||
default_ttl = dns.ttl.from_text(default_ttl)
|
||||
if rdclass is not None:
|
||||
rdclass = dns.rdataclass.RdataClass.make(rdclass)
|
||||
else:
|
||||
rdclass = None
|
||||
default_rdclass = dns.rdataclass.RdataClass.make(default_rdclass)
|
||||
if rdtype is not None:
|
||||
rdtype = dns.rdatatype.RdataType.make(rdtype)
|
||||
else:
|
||||
rdtype = None
|
||||
manager = RRSetsReaderManager(origin, relativize, default_rdclass)
|
||||
with manager.writer(True) as txn:
|
||||
tok = dns.tokenizer.Tokenizer(text, "<input>", idna_codec=idna_codec)
|
||||
reader = Reader(
|
||||
tok,
|
||||
default_rdclass,
|
||||
txn,
|
||||
allow_directives=False,
|
||||
force_name=name,
|
||||
force_ttl=ttl,
|
||||
force_rdclass=rdclass,
|
||||
force_rdtype=rdtype,
|
||||
default_ttl=default_ttl,
|
||||
)
|
||||
reader.read()
|
||||
return manager.rrsets
|
||||
37
venv/Lib/site-packages/dns/zonetypes.py
Normal file
37
venv/Lib/site-packages/dns/zonetypes.py
Normal file
@@ -0,0 +1,37 @@
|
||||
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
|
||||
|
||||
"""Common zone-related types."""
|
||||
|
||||
# This is a separate file to avoid import circularity between dns.zone and
|
||||
# the implementation of the ZONEMD type.
|
||||
|
||||
import hashlib
|
||||
|
||||
import dns.enum
|
||||
|
||||
|
||||
class DigestScheme(dns.enum.IntEnum):
|
||||
"""ZONEMD Scheme"""
|
||||
|
||||
SIMPLE = 1
|
||||
|
||||
@classmethod
|
||||
def _maximum(cls):
|
||||
return 255
|
||||
|
||||
|
||||
class DigestHashAlgorithm(dns.enum.IntEnum):
|
||||
"""ZONEMD Hash Algorithm"""
|
||||
|
||||
SHA384 = 1
|
||||
SHA512 = 2
|
||||
|
||||
@classmethod
|
||||
def _maximum(cls):
|
||||
return 255
|
||||
|
||||
|
||||
_digest_hashers = {
|
||||
DigestHashAlgorithm.SHA384: hashlib.sha384,
|
||||
DigestHashAlgorithm.SHA512: hashlib.sha512,
|
||||
}
|
||||
Reference in New Issue
Block a user