mirror of https://github.com/morpheus65535/bazarr
625 lines
24 KiB
Python
625 lines
24 KiB
Python
# 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
|
|
|
|
import dns.exception
|
|
import dns.name
|
|
import dns.node
|
|
import dns.rdataclass
|
|
import dns.rdatatype
|
|
import dns.rdata
|
|
import dns.rdtypes.ANY.SOA
|
|
import dns.rrset
|
|
import dns.tokenizer
|
|
import dns.transaction
|
|
import dns.ttl
|
|
import dns.grange
|
|
|
|
|
|
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
|
|
|
|
|
|
class Reader:
|
|
|
|
"""Read a DNS zone file into a transaction."""
|
|
|
|
def __init__(self, tok, rdclass, txn, allow_include=False,
|
|
allow_directives=True, force_name=None,
|
|
force_ttl=None, force_rdclass=None, force_rdtype=None,
|
|
default_ttl=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 = []
|
|
self.current_file = None
|
|
self.allow_include = allow_include
|
|
self.allow_directives = 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 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:
|
|
if self.default_ttl_known:
|
|
ttl = self.default_ttl
|
|
elif self.last_ttl_known:
|
|
ttl = self.last_ttl
|
|
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")
|
|
|
|
# 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(
|
|
"unknown rdatatype '%s'" % 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(
|
|
"caught exception {}: {}".format(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.
|
|
self.default_ttl = rd.minimum
|
|
self.default_ttl_known = True
|
|
if ttl is None:
|
|
# if we didn't have a TTL on the SOA, set it!
|
|
ttl = 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):
|
|
# 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
|
|
g1 = is_generate1.match(side)
|
|
if g1:
|
|
mod, sign, offset, width, base = g1.groups()
|
|
if sign == '':
|
|
sign = '+'
|
|
g2 = is_generate2.match(side)
|
|
if g2:
|
|
mod, sign, offset = g2.groups()
|
|
if sign == '':
|
|
sign = '+'
|
|
width = 0
|
|
base = 'd'
|
|
g3 = is_generate3.match(side)
|
|
if g3:
|
|
mod, sign, offset, width = g3.groups()
|
|
if sign == '':
|
|
sign = '+'
|
|
base = 'd'
|
|
|
|
if not (g1 or g2 or g3):
|
|
mod = ''
|
|
sign = '+'
|
|
offset = 0
|
|
width = 0
|
|
base = 'd'
|
|
|
|
if base != 'd':
|
|
raise NotImplementedError()
|
|
|
|
return mod, sign, offset, width, 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
|
|
# 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("unknown rdatatype '%s'" %
|
|
token.value)
|
|
|
|
# rhs (required)
|
|
rhs = token.value
|
|
|
|
# The code currently only supports base 'd', so the last value
|
|
# in the tuple _parse_modify returns is ignored
|
|
lmod, lsign, loffset, lwidth, _ = self._parse_modify(lhs)
|
|
rmod, rsign, roffset, rwidth, _ = self._parse_modify(rhs)
|
|
for i in range(start, stop + 1, step):
|
|
# +1 because bind is inclusive and python is exclusive
|
|
|
|
if lsign == '+':
|
|
lindex = i + int(loffset)
|
|
elif lsign == '-':
|
|
lindex = i - int(loffset)
|
|
|
|
if rsign == '-':
|
|
rindex = i - int(roffset)
|
|
elif rsign == '+':
|
|
rindex = i + int(roffset)
|
|
|
|
lzfindex = str(lindex).zfill(int(lwidth))
|
|
rzfindex = str(rindex).zfill(int(rwidth))
|
|
|
|
name = lhs.replace('$%s' % (lmod), lzfindex)
|
|
rdata = rhs.replace('$%s' % (rmod), rzfindex)
|
|
|
|
self.last_name = dns.name.from_text(name, self.current_origin,
|
|
self.tok.idna_codec)
|
|
name = self.last_name
|
|
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("caught exception %s: %s" %
|
|
(str(ty), str(va)))
|
|
|
|
self.txn.add(name, ttl, rd)
|
|
|
|
def read(self):
|
|
"""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 self.allow_directives:
|
|
c = token.value.upper()
|
|
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' and self.allow_include:
|
|
token = self.tok.get()
|
|
filename = token.value
|
|
token = self.tok.get()
|
|
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, 'r')
|
|
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(
|
|
"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(
|
|
"%s:%d: %s" % (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)
|
|
|
|
def _set_origin(self, origin):
|
|
pass
|
|
|
|
|
|
class RRSetsReaderManager(dns.transaction.TransactionManager):
|
|
def __init__(self, origin=dns.name.root, relativize=False,
|
|
rdclass=dns.rdataclass.IN):
|
|
self.origin = origin
|
|
self.relativize = relativize
|
|
self.rdclass = rdclass
|
|
self.rrsets = []
|
|
|
|
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):
|
|
self.rrsets = rrsets
|
|
|
|
|
|
def read_rrsets(text, name=None, ttl=None, rdclass=dns.rdataclass.IN,
|
|
default_rdclass=dns.rdataclass.IN,
|
|
rdtype=None, default_ttl=None, idna_codec=None,
|
|
origin=dns.name.root, relativize=False):
|
|
"""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)
|
|
default_rdclass = dns.rdataclass.RdataClass.make(default_rdclass)
|
|
if rdtype is not None:
|
|
rdtype = dns.rdatatype.RdataType.make(rdtype)
|
|
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
|