Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/zonefile.py: 11%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
3# Copyright (C) 2003-2007, 2009-2011 Nominum, Inc.
4#
5# Permission to use, copy, modify, and distribute this software and its
6# documentation for any purpose with or without fee is hereby granted,
7# provided that the above copyright notice and this permission notice
8# appear in all copies.
9#
10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
18"""DNS Zones."""
20import re
21import sys
22from collections.abc import Iterable
23from typing import Any, cast
25import dns.exception
26import dns.grange
27import dns.name
28import dns.node
29import dns.rdata
30import dns.rdataclass
31import dns.rdatatype
32import dns.rdtypes.ANY.SOA
33import dns.rrset
34import dns.tokenizer
35import dns.transaction
36import dns.ttl
39class UnknownOrigin(dns.exception.DNSException):
40 """Unknown origin"""
43class CNAMEAndOtherData(dns.exception.DNSException):
44 """A node has a CNAME and other data"""
47def _check_cname_and_other_data(txn, name, rdataset):
48 rdataset_kind = dns.node.NodeKind.classify_rdataset(rdataset)
49 node = txn.get_node(name)
50 if node is None:
51 # empty nodes are neutral.
52 return
53 node_kind = node.classify()
54 if (
55 node_kind == dns.node.NodeKind.CNAME
56 and rdataset_kind == dns.node.NodeKind.REGULAR
57 ):
58 raise CNAMEAndOtherData("rdataset type is not compatible with a CNAME node")
59 elif (
60 node_kind == dns.node.NodeKind.REGULAR
61 and rdataset_kind == dns.node.NodeKind.CNAME
62 ):
63 raise CNAMEAndOtherData(
64 "CNAME rdataset is not compatible with a regular data node"
65 )
66 # Otherwise at least one of the node and the rdataset is neutral, so
67 # adding the rdataset is ok
70SavedStateType = tuple[
71 dns.tokenizer.Tokenizer,
72 dns.name.Name | None, # current_origin
73 dns.name.Name | None, # last_name
74 Any | None, # current_file
75 int, # last_ttl
76 bool, # last_ttl_known
77 int, # default_ttl
78 bool, # default_ttl_known
79 bool, # default_ttl_from_soa
80]
83def _upper_dollarize(s):
84 s = s.upper()
85 if not s.startswith("$"):
86 s = "$" + s
87 return s
90class Reader:
91 """Read a DNS zone file into a transaction."""
93 def __init__(
94 self,
95 tok: dns.tokenizer.Tokenizer,
96 rdclass: dns.rdataclass.RdataClass,
97 txn: dns.transaction.Transaction,
98 allow_include: bool = False,
99 allow_directives: bool | Iterable[str] = True,
100 force_name: dns.name.Name | None = None,
101 force_ttl: int | None = None,
102 force_rdclass: dns.rdataclass.RdataClass | None = None,
103 force_rdtype: dns.rdatatype.RdataType | None = None,
104 default_ttl: int | None = None,
105 rfc2308_ttl: bool = True,
106 ):
107 self.tok = tok
108 self.zone_origin, self.relativize, _ = txn.manager.origin_information()
109 self.current_origin = self.zone_origin
110 self.last_ttl = 0
111 self.last_ttl_known = False
112 self.rfc2308_ttl = rfc2308_ttl
113 self.default_ttl_from_soa = False
114 if force_ttl is not None:
115 default_ttl = force_ttl
116 if default_ttl is None:
117 self.default_ttl = 0
118 self.default_ttl_known = False
119 else:
120 self.default_ttl = default_ttl
121 self.default_ttl_known = True
122 self.last_name = self.current_origin
123 self.zone_rdclass = rdclass
124 self.txn = txn
125 self.saved_state: list[SavedStateType] = []
126 self.current_file: Any | None = None
127 self.allowed_directives: set[str]
128 if allow_directives is True:
129 self.allowed_directives = {"$GENERATE", "$ORIGIN", "$TTL", "$UNICODE"}
130 if allow_include:
131 self.allowed_directives.add("$INCLUDE")
132 elif allow_directives is False:
133 # allow_include was ignored in earlier releases if allow_directives was
134 # False, so we continue that.
135 self.allowed_directives = set()
136 else:
137 # Note that if directives are explicitly specified, then allow_include
138 # is ignored.
139 self.allowed_directives = set(_upper_dollarize(d) for d in allow_directives)
140 self.force_name = force_name
141 self.force_ttl = force_ttl
142 self.force_rdclass = force_rdclass
143 self.force_rdtype = force_rdtype
144 self.txn.check_put_rdataset(_check_cname_and_other_data)
146 def _eat_line(self):
147 while 1:
148 token = self.tok.get()
149 if token.is_eol_or_eof():
150 break
152 def _get_identifier(self):
153 token = self.tok.get()
154 if not token.is_identifier():
155 raise dns.exception.SyntaxError
156 return token
158 def _implicit_ttl(self) -> int | None:
159 """Return the TTL to use for a record which did not specify one, or
160 ``None`` if no TTL can be determined yet.
162 An explicit default (from ``$TTL``, or from the *default_ttl* /
163 *force_ttl* constructor parameters) always wins. Otherwise, in
164 ``rfc2308_ttl`` mode, the most recently stated TTL (RFC 1035 section
165 5.1) is preferred over a default synthesized from the SOA MINIMUM
166 field, whose overloaded "default TTL" meaning RFC 2308 section 4
167 removed. In legacy mode the SOA-derived default wins, as it did
168 before RFC 2308.
169 """
170 if self.last_ttl_known and ( # only if the default doesn't take precedence
171 not self.default_ttl_known
172 or (self.rfc2308_ttl and self.default_ttl_from_soa)
173 ):
174 return self.last_ttl
175 if self.default_ttl_known:
176 return self.default_ttl
177 return None
179 def _rr_line(self) -> None:
180 """Process one line from a DNS zone file."""
181 token: dns.tokenizer.Token
182 name: dns.name.Name | None
183 # Name
184 if self.force_name is not None:
185 name = self.force_name
186 else:
187 if self.current_origin is None:
188 raise UnknownOrigin
189 token = self.tok.get(want_leading=True)
190 if not token.is_whitespace():
191 self.last_name = self.tok.as_name(token, self.current_origin)
192 else:
193 token = self.tok.get()
194 if token.is_eol_or_eof():
195 # treat leading WS followed by EOL/EOF as if they were EOL/EOF.
196 return
197 self.tok.unget(token)
198 name = self.last_name
199 if name is None:
200 raise dns.exception.SyntaxError("the last used name is undefined")
201 assert self.zone_origin is not None
202 if not name.is_subdomain(self.zone_origin):
203 self._eat_line()
204 return
205 if self.relativize:
206 name = name.relativize(self.zone_origin)
208 # TTL
209 if self.force_ttl is not None:
210 ttl = self.force_ttl
211 self.last_ttl = ttl
212 self.last_ttl_known = True
213 else:
214 token = self._get_identifier()
215 ttl = None
216 try:
217 ttl = dns.ttl.from_text(token.value)
218 self.last_ttl = ttl
219 self.last_ttl_known = True
220 except dns.ttl.BadTTL:
221 self.tok.unget(token)
223 # Class
224 if self.force_rdclass is not None:
225 rdclass = self.force_rdclass
226 else:
227 token = self._get_identifier()
228 try:
229 rdclass = dns.rdataclass.from_text(token.value)
230 except dns.exception.SyntaxError:
231 raise
232 except Exception:
233 rdclass = self.zone_rdclass
234 self.tok.unget(token)
235 if rdclass != self.zone_rdclass:
236 raise dns.exception.SyntaxError("RR class is not zone's class")
238 if ttl is None:
239 # support for <class> <ttl> <type> syntax
240 token = self._get_identifier()
241 try:
242 ttl = dns.ttl.from_text(token.value)
243 self.last_ttl = ttl
244 self.last_ttl_known = True
245 except dns.ttl.BadTTL:
246 ttl = self._implicit_ttl()
247 self.tok.unget(token)
249 # Type
250 if self.force_rdtype is not None:
251 rdtype = self.force_rdtype
252 else:
253 token = self._get_identifier()
254 try:
255 rdtype = dns.rdatatype.from_text(token.value)
256 except Exception:
257 raise dns.exception.SyntaxError(f"unknown rdatatype '{token.value}'")
259 try:
260 rd = dns.rdata.from_text(
261 rdclass,
262 rdtype,
263 self.tok,
264 self.current_origin,
265 self.relativize,
266 self.zone_origin,
267 )
268 except dns.exception.SyntaxError:
269 # Catch and reraise.
270 raise
271 except Exception:
272 # All exceptions that occur in the processing of rdata
273 # are treated as syntax errors. This is not strictly
274 # correct, but it is correct almost all of the time.
275 # We convert them to syntax errors so that we can emit
276 # helpful filename:line info.
277 ty, va = sys.exc_info()[:2]
278 raise dns.exception.SyntaxError(f"caught exception {str(ty)}: {str(va)}")
280 if not self.default_ttl_known and rdtype == dns.rdatatype.SOA:
281 # The pre-RFC2308 and pre-BIND9 behavior inherits the zone default
282 # TTL from the SOA minttl if no $TTL statement is present before the
283 # SOA is parsed.
284 soa_rd = cast(dns.rdtypes.ANY.SOA.SOA, rd)
285 self.default_ttl = soa_rd.minimum
286 self.default_ttl_known = True
287 self.default_ttl_from_soa = True
288 if ttl is None:
289 # if we didn't have a TTL on the SOA, set it!
290 ttl = self._implicit_ttl()
292 # TTL check. We had to wait until now to do this as the SOA RR's
293 # own TTL can be inferred from its minimum.
294 if ttl is None:
295 raise dns.exception.SyntaxError("Missing default TTL value")
297 self.txn.add(name, ttl, rd)
299 def _parse_modify(self, side: str) -> tuple[str, str, int, int, str]:
300 # Here we catch everything in '{' '}' in a group so we can replace it
301 # with ''.
302 is_generate1 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+),(.)}).*$")
303 is_generate2 = re.compile(r"^.*\$({(\+|-?)(\d+)}).*$")
304 is_generate3 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+)}).*$")
305 # Sometimes there are modifiers in the hostname. These come after
306 # the dollar sign. They are in the form: ${offset[,width[,base]]}.
307 # Make names
308 mod = ""
309 sign = "+"
310 offset = "0"
311 width = "0"
312 base = "d"
313 g1 = is_generate1.match(side)
314 if g1:
315 mod, sign, offset, width, base = g1.groups()
316 if sign == "":
317 sign = "+"
318 else:
319 g2 = is_generate2.match(side)
320 if g2:
321 mod, sign, offset = g2.groups()
322 if sign == "":
323 sign = "+"
324 width = "0"
325 base = "d"
326 else:
327 g3 = is_generate3.match(side)
328 if g3:
329 mod, sign, offset, width = g3.groups()
330 if sign == "":
331 sign = "+"
332 base = "d"
334 ioffset = int(offset)
335 iwidth = int(width)
337 if sign not in ["+", "-"]:
338 raise dns.exception.SyntaxError(f"invalid offset sign {sign}")
339 if base not in ["d", "o", "x", "X", "n", "N"]:
340 raise dns.exception.SyntaxError(f"invalid type {base}")
342 return mod, sign, ioffset, iwidth, base
344 def _generate_line(self):
345 # range lhs [ttl] [class] type rhs [ comment ]
346 """Process one line containing the GENERATE statement from a DNS
347 zone file."""
348 if self.current_origin is None:
349 raise UnknownOrigin
351 token = self.tok.get()
352 # Range (required)
353 try:
354 start, stop, step = dns.grange.from_text(token.value)
355 token = self.tok.get()
356 if not token.is_identifier():
357 raise dns.exception.SyntaxError
358 except Exception:
359 raise dns.exception.SyntaxError
361 # lhs (required)
362 try:
363 lhs = token.value
364 token = self.tok.get()
365 if not token.is_identifier():
366 raise dns.exception.SyntaxError
367 except Exception:
368 raise dns.exception.SyntaxError
370 # TTL
371 try:
372 ttl = dns.ttl.from_text(token.value)
373 self.last_ttl = ttl
374 self.last_ttl_known = True
375 token = self.tok.get()
376 if not token.is_identifier():
377 raise dns.exception.SyntaxError
378 except dns.ttl.BadTTL:
379 # We don't go to the extra "look at the SOA" level of effort for
380 # $GENERATE, because the user really ought to have defined a TTL
381 # somehow!
382 ttl = self._implicit_ttl()
383 if ttl is None:
384 raise dns.exception.SyntaxError("Missing default TTL value")
386 # Class
387 try:
388 rdclass = dns.rdataclass.from_text(token.value)
389 token = self.tok.get()
390 if not token.is_identifier():
391 raise dns.exception.SyntaxError
392 except dns.exception.SyntaxError:
393 raise dns.exception.SyntaxError
394 except Exception:
395 rdclass = self.zone_rdclass
396 if rdclass != self.zone_rdclass:
397 raise dns.exception.SyntaxError("RR class is not zone's class")
398 # Type
399 try:
400 rdtype = dns.rdatatype.from_text(token.value)
401 token = self.tok.get()
402 if not token.is_identifier():
403 raise dns.exception.SyntaxError
404 except Exception:
405 raise dns.exception.SyntaxError(f"unknown rdatatype '{token.value}'")
407 # rhs (required)
408 rhs = token.value
410 def _calculate_index(counter: int, offset_sign: str, offset: int) -> int:
411 """Calculate the index from the counter and offset."""
412 if offset_sign == "-":
413 offset *= -1
414 return counter + offset
416 def _format_index(index: int, base: str, width: int) -> str:
417 """Format the index with the given base, and zero-fill it
418 to the given width."""
419 if base in ["d", "o", "x", "X"]:
420 return format(index, base).zfill(width)
422 # base can only be n or N here
423 hexa = _format_index(index, "x", width)
424 nibbles = ".".join(hexa[::-1])[:width]
425 if base == "N":
426 nibbles = nibbles.upper()
427 return nibbles
429 lmod, lsign, loffset, lwidth, lbase = self._parse_modify(lhs)
430 rmod, rsign, roffset, rwidth, rbase = self._parse_modify(rhs)
431 for i in range(start, stop + 1, step):
432 # +1 because bind is inclusive and python is exclusive
434 lindex = _calculate_index(i, lsign, loffset)
435 rindex = _calculate_index(i, rsign, roffset)
437 lzfindex = _format_index(lindex, lbase, lwidth)
438 rzfindex = _format_index(rindex, rbase, rwidth)
440 name = lhs.replace(f"${lmod}", lzfindex)
441 rdata = rhs.replace(f"${rmod}", rzfindex)
443 self.last_name = dns.name.from_text(
444 name, self.current_origin, self.tok.idna_codec
445 )
446 name = self.last_name
447 assert self.zone_origin is not None
448 if not name.is_subdomain(self.zone_origin):
449 self._eat_line()
450 return
451 if self.relativize:
452 name = name.relativize(self.zone_origin)
454 try:
455 rd = dns.rdata.from_text(
456 rdclass,
457 rdtype,
458 rdata,
459 self.current_origin,
460 self.relativize,
461 self.zone_origin,
462 )
463 except dns.exception.SyntaxError:
464 # Catch and reraise.
465 raise
466 except Exception:
467 # All exceptions that occur in the processing of rdata
468 # are treated as syntax errors. This is not strictly
469 # correct, but it is correct almost all of the time.
470 # We convert them to syntax errors so that we can emit
471 # helpful filename:line info.
472 ty, va = sys.exc_info()[:2]
473 raise dns.exception.SyntaxError(
474 f"caught exception {str(ty)}: {str(va)}"
475 )
477 self.txn.add(name, ttl, rd)
479 def read(self) -> None:
480 """Read a DNS zone file and build a zone object.
482 :raises dns.zone.NoSOA: if there is no SOA RR at the zone origin.
483 :raises dns.zone.NoNS: if there is no NS RRset at the zone origin.
484 """
486 try:
487 while 1:
488 token = self.tok.get(True, True)
489 if token.is_eof():
490 if self.current_file is not None:
491 self.current_file.close()
492 if len(self.saved_state) > 0:
493 (
494 self.tok,
495 self.current_origin,
496 self.last_name,
497 self.current_file,
498 self.last_ttl,
499 self.last_ttl_known,
500 self.default_ttl,
501 self.default_ttl_known,
502 self.default_ttl_from_soa,
503 ) = self.saved_state.pop(-1)
504 continue
505 break
506 elif token.is_eol():
507 continue
508 elif token.is_comment():
509 self.tok.get_eol()
510 continue
511 elif (
512 token.is_identifier()
513 and token.value[0] == "$"
514 and len(self.allowed_directives) > 0
515 ):
516 # Note that we only run directive processing code if at least
517 # one directive is allowed in order to be backwards compatible
518 c = token.value.upper()
519 if c not in self.allowed_directives:
520 raise dns.exception.SyntaxError(
521 f"zone file directive '{c}' is not allowed"
522 )
523 if c == "$TTL":
524 token = self.tok.get()
525 if not token.is_identifier():
526 raise dns.exception.SyntaxError("bad $TTL")
527 self.default_ttl = dns.ttl.from_text(token.value)
528 self.default_ttl_known = True
529 self.default_ttl_from_soa = False
530 self.tok.get_eol()
531 elif c == "$ORIGIN":
532 # A relative $ORIGIN is relative to the current origin
533 # (RFC 1035 section 5.1).
534 origin = self.tok.get_name(self.current_origin)
535 if not origin.is_absolute():
536 raise UnknownOrigin
537 self.current_origin = origin
538 self.tok.get_eol()
539 if self.zone_origin is None:
540 self.zone_origin = self.current_origin
541 self.txn._set_origin(self.current_origin)
542 elif c == "$INCLUDE":
543 token = self.tok.get()
544 filename = token.value
545 token = self.tok.get()
546 new_origin: dns.name.Name | None
547 if token.is_identifier():
548 new_origin = dns.name.from_text(
549 token.value, self.current_origin, self.tok.idna_codec
550 )
551 self.tok.get_eol()
552 elif not token.is_eol_or_eof():
553 raise dns.exception.SyntaxError("bad origin in $INCLUDE")
554 else:
555 new_origin = self.current_origin
556 self.saved_state.append(
557 (
558 self.tok,
559 self.current_origin,
560 self.last_name,
561 self.current_file,
562 self.last_ttl,
563 self.last_ttl_known,
564 self.default_ttl,
565 self.default_ttl_known,
566 self.default_ttl_from_soa,
567 )
568 )
569 self.current_file = open(filename, encoding="utf-8")
570 self.tok = dns.tokenizer.Tokenizer(self.current_file, filename)
571 self.current_origin = new_origin
572 elif c == "$GENERATE":
573 self._generate_line()
574 elif c == "$UNICODE":
575 while True:
576 token = self.tok.get()
577 if token.is_eol_or_eof():
578 break
579 if not token.is_identifier():
580 raise dns.exception.SyntaxError("bad $UNICODE")
581 self.txn.add_unicode(token.value)
582 if token.value == "2008":
583 self.tok.idna_codec = dns.name.IDNA_2008_Practical
584 elif token.value == "2003":
585 self.tok.idna_codec = dns.name.IDNA_2003_Practical
586 else:
587 raise dns.exception.SyntaxError(
588 f"Unknown zone file directive '{c}'"
589 )
590 continue
591 self.tok.unget(token)
592 self._rr_line()
593 except dns.exception.SyntaxError as detail:
594 filename, line_number = self.tok.where()
595 if detail is None:
596 detail = "syntax error"
597 ex = dns.exception.SyntaxError(f"{filename}:{line_number}: {detail}")
598 tb = sys.exc_info()[2]
599 raise ex.with_traceback(tb) from None
602class RRsetsReaderTransaction(dns.transaction.Transaction):
603 def __init__(self, manager, replacement, read_only):
604 assert not read_only
605 super().__init__(manager, replacement, read_only)
606 self.rdatasets = {}
608 def _get_rdataset(self, name, rdtype, covers):
609 return self.rdatasets.get((name, rdtype, covers))
611 def _get_node(self, name):
612 rdatasets = []
613 for (rdataset_name, _, _), rdataset in self.rdatasets.items():
614 if name == rdataset_name:
615 rdatasets.append(rdataset)
616 if len(rdatasets) == 0:
617 return None
618 node = dns.node.Node()
619 node.rdatasets = rdatasets
620 return node
622 def _put_rdataset(self, name, rdataset):
623 self.rdatasets[(name, rdataset.rdtype, rdataset.covers)] = rdataset
625 def _delete_name(self, name):
626 # First remove any changes involving the name
627 remove = []
628 for key in self.rdatasets:
629 if key[0] == name:
630 remove.append(key)
631 if len(remove) > 0:
632 for key in remove:
633 del self.rdatasets[key]
635 def _delete_rdataset(self, name, rdtype, covers):
636 try:
637 del self.rdatasets[(name, rdtype, covers)]
638 except KeyError:
639 pass
641 def _name_exists(self, name):
642 for n, _, _ in self.rdatasets:
643 if n == name:
644 return True
645 return False
647 def _changed(self):
648 return len(self.rdatasets) > 0
650 def _end_transaction(self, commit):
651 if commit and self._changed():
652 rrsets = []
653 for (name, _, _), rdataset in self.rdatasets.items():
654 rrset = dns.rrset.RRset(
655 name, rdataset.rdclass, rdataset.rdtype, rdataset.covers
656 )
657 rrset.update(rdataset)
658 rrsets.append(rrset)
659 manager = cast(RRSetsReaderManager, self.manager)
660 manager.set_rrsets(rrsets)
662 def _set_origin(self, origin):
663 pass
665 def _iterate_rdatasets(self):
666 raise NotImplementedError # pragma: no cover
668 def _iterate_names(self):
669 raise NotImplementedError # pragma: no cover
672class RRSetsReaderManager(dns.transaction.TransactionManager):
673 def __init__(
674 self,
675 origin: dns.name.Name | None = dns.name.root,
676 relativize: bool = False,
677 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
678 ):
679 self.origin = origin
680 self.relativize = relativize
681 self.rdclass = rdclass
682 self.rrsets: list[dns.rrset.RRset] = []
684 def reader(self): # pragma: no cover
685 raise NotImplementedError
687 def writer(self, replacement=False):
688 assert replacement is True
689 return RRsetsReaderTransaction(self, True, False)
691 def get_class(self):
692 return self.rdclass
694 def origin_information(self):
695 if self.relativize:
696 effective = dns.name.empty
697 else:
698 effective = self.origin
699 return (self.origin, self.relativize, effective)
701 def set_rrsets(self, rrsets: list[dns.rrset.RRset]) -> None:
702 self.rrsets = rrsets
705def read_rrsets(
706 text: Any,
707 name: dns.name.Name | str | None = None,
708 ttl: int | None = None,
709 rdclass: dns.rdataclass.RdataClass | str | None = dns.rdataclass.IN,
710 default_rdclass: dns.rdataclass.RdataClass | str = dns.rdataclass.IN,
711 rdtype: dns.rdatatype.RdataType | str | None = None,
712 default_ttl: int | str | None = None,
713 idna_codec: dns.name.IDNACodec | None = None,
714 origin: dns.name.Name | str | None = dns.name.root,
715 relativize: bool = False,
716 rfc2308_ttl: bool = True,
717) -> list[dns.rrset.RRset]:
718 """Read one or more rrsets from the specified text, possibly subject
719 to restrictions.
721 :param text: The input to process, either a file object or a string.
722 :param name: The owner name of the rrset. If not ``None``, the owner
723 name is forced and the input must not specify one. If ``None``,
724 any owner names are allowed and must be present in the input.
725 :type name: str, :py:class:`dns.name.Name`, or ``None``
726 :param ttl: If not ``None``, the TTL is forced to the specified value
727 and the input must not specify a TTL. If ``None``, a TTL may be
728 specified in the input; if not specified, *default_ttl* will be used.
729 :type ttl: int, str, or ``None``
730 :param rdclass: If not ``None``, the class is forced to the specified
731 value and the input must not specify a class. If ``None``, the input
732 may optionally specify a class matching *default_rdclass*. Note that
733 rrsets with differing classes cannot be returned; ``None`` here simply
734 allows the user to optionally write a class in the input.
735 :type rdclass: :py:class:`dns.rdataclass.RdataClass`, str, or ``None``
736 :param default_rdclass: The class of the returned rrsets.
737 :type default_rdclass: :py:class:`dns.rdataclass.RdataClass` or str
738 :param rdtype: If not ``None``, the type is forced to the specified value
739 and the input must not specify a type. If ``None``, a type must be
740 present for each RR.
741 :type rdtype: :py:class:`dns.rdatatype.RdataType`, str, or ``None``
742 :param default_ttl: If not ``None``, and the TTL is not forced and not
743 specified in the input, this value will be used. If ``None``, an
744 error will occur if the TTL is not forced and not specified.
745 :type default_ttl: int, str, or ``None``
746 :param idna_codec: The IDNA encoder/decoder. If ``None``, the default
747 IDNA encoder/decoder is used. Note that codecs only apply to the
748 owner name; dnspython does not do IDNA for names in rdata.
749 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
750 :param origin: The origin for any relative names in the input, and also
751 the origin to relativize to if *relativize* is ``True``.
752 :type origin: str, :py:class:`dns.name.Name`, or ``None``
753 :param bool relativize: If ``True``, names are relativized to *origin*;
754 if ``False``, relative names in the input are made absolute by
755 appending *origin*.
756 :param bool rfc2308_ttl: If ``True``, a TTL-less record inherits the most
757 recently stated TTL in preference to a default derived from the SOA
758 MINIMUM field, per RFC 2308 section 4. The default is ``True``. If
759 ``False``, the historical pre-RFC 2308 behavior is used.
760 :rtype: list[:py:class:`dns.rrset.RRset`]
761 """
762 if isinstance(origin, str):
763 origin = dns.name.from_text(origin, dns.name.root, idna_codec)
764 if isinstance(name, str):
765 name = dns.name.from_text(name, origin, idna_codec)
766 if isinstance(ttl, str):
767 ttl = dns.ttl.from_text(ttl)
768 if isinstance(default_ttl, str):
769 default_ttl = dns.ttl.from_text(default_ttl)
770 if rdclass is not None:
771 rdclass = dns.rdataclass.RdataClass.make(rdclass)
772 else:
773 rdclass = None
774 default_rdclass = dns.rdataclass.RdataClass.make(default_rdclass)
775 if rdtype is not None:
776 rdtype = dns.rdatatype.RdataType.make(rdtype)
777 else:
778 rdtype = None
779 manager = RRSetsReaderManager(origin, relativize, default_rdclass)
780 with manager.writer(True) as txn:
781 tok = dns.tokenizer.Tokenizer(text, "<input>", idna_codec=idna_codec)
782 reader = Reader(
783 tok,
784 default_rdclass,
785 txn,
786 allow_directives=False,
787 force_name=name,
788 force_ttl=ttl,
789 force_rdclass=rdclass,
790 force_rdtype=rdtype,
791 default_ttl=default_ttl,
792 rfc2308_ttl=rfc2308_ttl,
793 )
794 reader.read()
795 return manager.rrsets