Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/zone.py: 23%
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 dataclasses
21import io
22import os
23import struct
24from collections.abc import Callable, Iterable, Iterator, MutableMapping
25from typing import Any, BinaryIO, TextIO, cast
27import dns._file_util
28import dns.exception
29import dns.immutable
30import dns.name
31import dns.node
32import dns.rdata
33import dns.rdataclass
34import dns.rdataset
35import dns.rdatatype
36import dns.rdtypes.ANY.SOA
37import dns.rdtypes.ANY.ZONEMD
38import dns.rrset
39import dns.tokenizer
40import dns.transaction
41import dns.zonefile
42from dns.zonetypes import DigestHashAlgorithm, DigestScheme, _digest_hashers
45class BadZone(dns.exception.DNSException):
46 """The DNS zone is malformed."""
49class NoSOA(BadZone):
50 """The DNS zone has no SOA RR at its origin."""
53class NoNS(BadZone):
54 """The DNS zone has no NS RRset at its origin."""
57class UnknownOrigin(BadZone):
58 """The DNS zone's origin is unknown."""
61class UnsupportedDigestScheme(dns.exception.DNSException):
62 """The zone digest's scheme is unsupported."""
65class UnsupportedDigestHashAlgorithm(dns.exception.DNSException):
66 """The zone digest's origin is unsupported."""
69class NoDigest(dns.exception.DNSException):
70 """The DNS zone has no ZONEMD RRset at its origin."""
73class DigestVerificationFailure(dns.exception.DNSException):
74 """The ZONEMD digest failed to verify."""
77def _validate_name(
78 name: dns.name.Name,
79 origin: dns.name.Name | None,
80 relativize: bool,
81) -> dns.name.Name:
82 # This name validation code is shared by Zone and Version
83 if origin is None:
84 # This should probably never happen as other code (e.g.
85 # _rr_line) will notice the lack of an origin before us, but
86 # we check just in case!
87 raise KeyError("no zone origin is defined")
88 if name.is_absolute():
89 if not name.is_subdomain(origin):
90 raise KeyError("name parameter must be a subdomain of the zone origin")
91 if relativize:
92 name = name.relativize(origin)
93 else:
94 # We have a relative name. Make sure that the derelativized name is
95 # not too long.
96 try:
97 abs_name = name.derelativize(origin)
98 except dns.name.NameTooLong:
99 # We map dns.name.NameTooLong to KeyError to be consistent with
100 # the other exceptions above.
101 raise KeyError("relative name too long for zone")
102 if not relativize:
103 # We have a relative name in a non-relative zone, so use the
104 # derelativized name.
105 name = abs_name
106 return name
109@dataclasses.dataclass(frozen=True)
110class ZoneStyle(dns.node.NodeStyle):
111 """Zone text styles.
113 A :py:class:`dns.zone.ZoneStyle` is also a :py:class:`dns.name.NameStyle`,
114 :py:class:`dns.rdata.RdataStyle`, :py:class:`dns.rdataset.RdatasetStyle`,
115 and :py:class:`dns.node.NodeStyle`.
116 See those classes for a description of their options.
118 :param bool sorted: If ``True`` (the default), the file will be written
119 with the names sorted in DNSSEC order from least to greatest.
120 Otherwise the names will be written in whatever order they happen
121 to have in the zone's dictionary.
122 :param nl: The end of line string. If ``None`` (the default), the
123 output will use the platform's native end-of-line marker (i.e. LF
124 on POSIX, CRLF on Windows).
125 :type nl: str or ``None``
126 :param bool want_origin: If ``True``, emit a ``$ORIGIN`` line at the
127 start of the output. If ``False`` (the default), do not emit one.
128 :param bool want_unicode_directive: If ``True`` and the zone has a
129 non-empty ``unicode`` attribute, then emit a ``$UNICODE`` line in
130 the output. This directive is not standard, but allows dnspython
131 and other aware software to read and write Unicode zonefiles without
132 changing the rendering of names and TXT-like records.
133 """
135 sorted: bool = True
136 nl: str | None = None
137 want_origin: bool = False
138 want_unicode_directive: bool = True
141class Zone(dns.transaction.TransactionManager):
142 """A DNS zone.
144 A ``Zone`` is a mapping from names to nodes. The zone object may be
145 treated like a Python dictionary, e.g. ``zone[name]`` will retrieve
146 the node associated with that name. The *name* may be a
147 ``dns.name.Name object``, or it may be a string. In either case,
148 if the name is relative it is treated as relative to the origin of
149 the zone.
150 """
152 node_factory: Callable[[], dns.node.Node] = dns.node.Node
153 map_factory: Callable[[], MutableMapping[dns.name.Name, dns.node.Node]] = dict
154 # We only require the version types as "Version" to allow for flexibility, as
155 # only the version protocol matters
156 writable_version_factory: Callable[["Zone", bool], "Version"] | None = None
157 immutable_version_factory: Callable[["Version"], "Version"] | None = None
159 __slots__ = ["rdclass", "origin", "nodes", "relativize"]
161 def __init__(
162 self,
163 origin: dns.name.Name | str | None,
164 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
165 relativize: bool = True,
166 ):
167 """Initialize a zone object.
169 :param origin: The origin of the zone. It may be a
170 :py:class:`dns.name.Name`, a ``str``, or ``None``. If ``None``,
171 then the zone's origin will be set by the first ``$ORIGIN`` line
172 in a zone file.
173 :type origin: :py:class:`dns.name.Name`, str, or ``None``
174 :param rdclass: The zone's rdata class; the default is class IN.
175 :type rdclass: :py:class:`dns.rdataclass.RdataClass`
176 :param bool relativize: Whether domain names are relativized to the
177 zone's origin. The default is ``True``.
178 """
180 if origin is not None:
181 if isinstance(origin, str):
182 origin = dns.name.from_text(origin)
183 elif not isinstance(origin, dns.name.Name):
184 raise ValueError("origin parameter must be convertible to a DNS name")
185 if not origin.is_absolute():
186 raise ValueError("origin parameter must be an absolute name")
187 self.origin: dns.name.Name | None = origin
188 self.rdclass = rdclass
189 self.nodes: MutableMapping[dns.name.Name, dns.node.Node] = self.map_factory()
190 self.relativize = relativize
191 self.unicode: set[str] = set()
193 def __eq__(self, other):
194 """Two zones are equal if they have the same origin, class, and
195 nodes.
197 :rtype: bool
198 """
200 if not isinstance(other, Zone):
201 return False
202 if (
203 self.rdclass != other.rdclass
204 or self.origin != other.origin
205 or self.nodes != other.nodes
206 ):
207 return False
208 return True
210 def __ne__(self, other):
211 """Are two zones not equal?
213 :rtype: bool
214 """
216 return not self.__eq__(other)
218 def _validate_name(self, name: dns.name.Name | str) -> dns.name.Name:
219 # Note that any changes in this method should have corresponding changes
220 # made in the Version _validate_name() method.
221 if isinstance(name, str):
222 name = dns.name.from_text(name, None)
223 elif not isinstance(name, dns.name.Name):
224 raise KeyError("name parameter must be convertible to a DNS name")
225 return _validate_name(name, self.origin, self.relativize)
227 def __getitem__(self, key):
228 key = self._validate_name(key)
229 return self.nodes[key]
231 def __setitem__(self, key, value):
232 key = self._validate_name(key)
233 self.nodes[key] = value
235 def __delitem__(self, key):
236 key = self._validate_name(key)
237 del self.nodes[key]
239 def __iter__(self):
240 return self.nodes.__iter__()
242 def keys(self):
243 return self.nodes.keys()
245 def values(self):
246 return self.nodes.values()
248 def items(self):
249 return self.nodes.items()
251 def get(self, key):
252 key = self._validate_name(key)
253 return self.nodes.get(key)
255 def __contains__(self, key):
256 key = self._validate_name(key)
257 return key in self.nodes
259 def find_node(
260 self, name: dns.name.Name | str, create: bool = False
261 ) -> dns.node.Node:
262 """Find a node in the zone, possibly creating it.
264 :param name: The name of the node to find. The value may be a
265 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
266 must be a subdomain of the zone's origin. If ``zone.relativize``
267 is ``True``, then the name will be relativized.
268 :type name: :py:class:`dns.name.Name` or str
269 :param bool create: If ``True``, the node will be created if it does
270 not exist.
271 :raises KeyError: if the name is not known and *create* was not
272 specified, or if the name was not a subdomain of the origin.
273 :rtype: :py:class:`dns.node.Node`
274 """
276 name = self._validate_name(name)
277 node = self.nodes.get(name)
278 if node is None:
279 if not create:
280 raise KeyError
281 node = self.node_factory()
282 self.nodes[name] = node
283 return node
285 def get_node(
286 self, name: dns.name.Name | str, create: bool = False
287 ) -> dns.node.Node | None:
288 """Get a node in the zone, possibly creating it.
290 This method is like :py:meth:`find_node`, except it returns ``None``
291 instead of raising an exception if the node does not exist and
292 creation has not been requested.
294 :param name: The name of the node to find. The value may be a
295 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
296 must be a subdomain of the zone's origin. If ``zone.relativize``
297 is ``True``, then the name will be relativized.
298 :type name: :py:class:`dns.name.Name` or str
299 :param bool create: If ``True``, the node will be created if it does
300 not exist.
301 :rtype: :py:class:`dns.node.Node` or ``None``
302 """
304 try:
305 node = self.find_node(name, create)
306 except KeyError:
307 node = None
308 return node
310 def delete_node(self, name: dns.name.Name | str) -> None:
311 """Delete the specified node if it exists.
313 It is not an error if the node does not exist.
315 :param name: The name of the node to delete. The value may be a
316 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
317 must be a subdomain of the zone's origin. If ``zone.relativize``
318 is ``True``, then the name will be relativized.
319 :type name: :py:class:`dns.name.Name` or str
320 """
322 name = self._validate_name(name)
323 if name in self.nodes:
324 del self.nodes[name]
326 def find_rdataset(
327 self,
328 name: dns.name.Name | str,
329 rdtype: dns.rdatatype.RdataType | str,
330 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
331 create: bool = False,
332 ) -> dns.rdataset.Rdataset:
333 """Look for an rdataset with the specified name and type in the zone,
334 and return an rdataset encapsulating it.
336 The rdataset returned is not a copy; changes to it will change
337 the zone.
339 :param name: The name of the node to find. The value may be a
340 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
341 must be a subdomain of the zone's origin. If ``zone.relativize``
342 is ``True``, then the name will be relativized.
343 :type name: :py:class:`dns.name.Name` or str
344 :param rdtype: The rdata type desired.
345 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
346 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
347 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
348 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
349 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
350 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
351 :type covers: :py:class:`dns.rdatatype.RdataType` or str
352 :param bool create: If ``True``, the node will be created if it does
353 not exist.
354 :raises KeyError: if the name or type are not found and *create* was
355 not specified, or if the name was not a subdomain of the origin.
356 :rtype: :py:class:`dns.rdataset.Rdataset`
357 """
359 name = self._validate_name(name)
360 rdtype = dns.rdatatype.RdataType.make(rdtype)
361 covers = dns.rdatatype.RdataType.make(covers)
362 node = self.find_node(name, create)
363 return node.find_rdataset(self.rdclass, rdtype, covers, create)
365 def get_rdataset(
366 self,
367 name: dns.name.Name | str,
368 rdtype: dns.rdatatype.RdataType | str,
369 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
370 create: bool = False,
371 ) -> dns.rdataset.Rdataset | None:
372 """Look for an rdataset with the specified name and type in the zone.
374 This method is like :py:meth:`find_rdataset`, except it returns
375 ``None`` instead of raising an exception if the rdataset does not
376 exist and creation has not been requested.
378 The rdataset returned is not a copy; changes to it will change
379 the zone.
381 :param name: The name of the node to find. The value may be a
382 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
383 must be a subdomain of the zone's origin. If ``zone.relativize``
384 is ``True``, then the name will be relativized.
385 :type name: :py:class:`dns.name.Name` or str
386 :param rdtype: The rdata type desired.
387 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
388 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
389 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
390 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
391 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
392 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
393 :type covers: :py:class:`dns.rdatatype.RdataType` or str
394 :param bool create: If ``True``, the node will be created if it does
395 not exist.
396 :rtype: :py:class:`dns.rdataset.Rdataset` or ``None``
397 """
399 try:
400 rdataset = self.find_rdataset(name, rdtype, covers, create)
401 except KeyError:
402 rdataset = None
403 return rdataset
405 def delete_rdataset(
406 self,
407 name: dns.name.Name | str,
408 rdtype: dns.rdatatype.RdataType | str,
409 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
410 ) -> None:
411 """Delete the rdataset matching *rdtype* and *covers*, if it
412 exists at the node specified by *name*.
414 It is not an error if the node does not exist, or if there is no
415 matching rdataset at the node. If the node has no rdatasets after
416 the deletion, it will itself be deleted.
418 :param name: The name of the node. The value may be a
419 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
420 must be a subdomain of the zone's origin. If ``zone.relativize``
421 is ``True``, then the name will be relativized.
422 :type name: :py:class:`dns.name.Name` or str
423 :param rdtype: The rdata type desired.
424 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
425 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
426 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
427 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
428 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
429 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
430 :type covers: :py:class:`dns.rdatatype.RdataType` or str
431 """
433 name = self._validate_name(name)
434 rdtype = dns.rdatatype.RdataType.make(rdtype)
435 covers = dns.rdatatype.RdataType.make(covers)
436 node = self.get_node(name)
437 if node is not None:
438 node.delete_rdataset(self.rdclass, rdtype, covers)
439 if len(node) == 0:
440 self.delete_node(name)
442 def replace_rdataset(
443 self, name: dns.name.Name | str, replacement: dns.rdataset.Rdataset
444 ) -> None:
445 """Replace an rdataset at name.
447 It is not an error if there is no rdataset matching *replacement*.
449 Ownership of the *replacement* object is transferred to the zone;
450 in other words, this method does not store a copy of *replacement*
451 at the node, it stores *replacement* itself. If the node does not
452 exist, it is created.
454 :param name: The name of the node. The value may be a
455 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
456 must be a subdomain of the zone's origin. If ``zone.relativize``
457 is ``True``, then the name will be relativized.
458 :type name: :py:class:`dns.name.Name` or str
459 :param replacement: The replacement rdataset.
460 :type replacement: :py:class:`dns.rdataset.Rdataset`
461 """
463 if replacement.rdclass != self.rdclass:
464 raise ValueError("replacement.rdclass != zone.rdclass")
465 node = self.find_node(name, True)
466 node.replace_rdataset(replacement)
468 def find_rrset(
469 self,
470 name: dns.name.Name | str,
471 rdtype: dns.rdatatype.RdataType | str,
472 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
473 ) -> dns.rrset.RRset:
474 """Look for an rdataset with the specified name and type in the zone,
475 and return an RRset encapsulating it.
477 This method is less efficient than the similar
478 :py:meth:`find_rdataset` because it creates an RRset instead of
479 returning the matching rdataset. It may be more convenient for some
480 uses since it returns an object which binds the owner name to the
481 rdataset.
483 This method may not be used to create new nodes or rdatasets;
484 use :py:meth:`find_rdataset` instead.
486 :param name: The name of the node to find. The value may be a
487 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
488 must be a subdomain of the zone's origin. If ``zone.relativize``
489 is ``True``, then the name will be relativized.
490 :type name: :py:class:`dns.name.Name` or str
491 :param rdtype: The rdata type desired.
492 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
493 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
494 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
495 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
496 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
497 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
498 :type covers: :py:class:`dns.rdatatype.RdataType` or str
499 :raises KeyError: if the name or type are not found, or if the name
500 was not a subdomain of the origin.
501 :rtype: :py:class:`dns.rrset.RRset`
502 """
504 vname = self._validate_name(name)
505 rdtype = dns.rdatatype.RdataType.make(rdtype)
506 covers = dns.rdatatype.RdataType.make(covers)
507 rdataset = self.nodes[vname].find_rdataset(self.rdclass, rdtype, covers)
508 rrset = dns.rrset.RRset(vname, self.rdclass, rdtype, covers)
509 rrset.update(rdataset)
510 return rrset
512 def get_rrset(
513 self,
514 name: dns.name.Name | str,
515 rdtype: dns.rdatatype.RdataType | str,
516 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
517 ) -> dns.rrset.RRset | None:
518 """Look for an rdataset with the specified name and type in the zone,
519 and return an RRset encapsulating it.
521 This method is less efficient than the similar :py:meth:`get_rdataset`
522 because it creates an RRset instead of returning the matching
523 rdataset. It may be more convenient for some uses since it returns
524 an object which binds the owner name to the rdataset.
526 This method may not be used to create new nodes or rdatasets;
527 use :py:meth:`get_rdataset` instead.
529 :param name: The name of the node to find. The value may be a
530 :py:class:`dns.name.Name` or a ``str``. If absolute, the name
531 must be a subdomain of the zone's origin. If ``zone.relativize``
532 is ``True``, then the name will be relativized.
533 :type name: :py:class:`dns.name.Name` or str
534 :param rdtype: The rdata type desired.
535 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
536 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
537 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
538 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
539 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
540 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
541 :type covers: :py:class:`dns.rdatatype.RdataType` or str
542 :rtype: :py:class:`dns.rrset.RRset` or ``None``
543 """
545 try:
546 rrset = self.find_rrset(name, rdtype, covers)
547 except KeyError:
548 rrset = None
549 return rrset
551 def iterate_rdatasets(
552 self,
553 rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.ANY,
554 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
555 ) -> Iterator[tuple[dns.name.Name, dns.rdataset.Rdataset]]:
556 """Return a generator which yields ``(name, rdataset)`` tuples for
557 all rdatasets in the zone which have the specified *rdtype* and
558 *covers*. If *rdtype* is :py:attr:`dns.rdatatype.ANY` (the default),
559 then all rdatasets will be matched.
561 :param rdtype: The rdata type desired.
562 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
563 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
564 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
565 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
566 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
567 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
568 :type covers: :py:class:`dns.rdatatype.RdataType` or str
569 """
571 rdtype = dns.rdatatype.RdataType.make(rdtype)
572 covers = dns.rdatatype.RdataType.make(covers)
573 for name, node in self.items():
574 for rds in node:
575 if rdtype == dns.rdatatype.ANY or (
576 rds.rdtype == rdtype and rds.covers == covers
577 ):
578 yield (name, rds)
580 def iterate_rdatas(
581 self,
582 rdtype: dns.rdatatype.RdataType | str = dns.rdatatype.ANY,
583 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE,
584 ) -> Iterator[tuple[dns.name.Name, int, dns.rdata.Rdata]]:
585 """Return a generator which yields ``(name, ttl, rdata)`` tuples for
586 all rdatas in the zone which have the specified *rdtype* and *covers*.
587 If *rdtype* is :py:attr:`dns.rdatatype.ANY` (the default), then all
588 rdatas will be matched.
590 :param rdtype: The rdata type desired.
591 :type rdtype: :py:class:`dns.rdatatype.RdataType` or str
592 :param covers: The covered type. Usually :py:attr:`dns.rdatatype.NONE`,
593 but if *rdtype* is :py:attr:`dns.rdatatype.SIG` or
594 :py:attr:`dns.rdatatype.RRSIG`, then this is the rdata type the
595 SIG/RRSIG covers. The library treats the SIG and RRSIG types as
596 a family of types, e.g. RRSIG(A), RRSIG(NS), RRSIG(SOA).
597 :type covers: :py:class:`dns.rdatatype.RdataType` or str
598 """
600 rdtype = dns.rdatatype.RdataType.make(rdtype)
601 covers = dns.rdatatype.RdataType.make(covers)
602 for name, node in self.items():
603 for rds in node:
604 if rdtype == dns.rdatatype.ANY or (
605 rds.rdtype == rdtype and rds.covers == covers
606 ):
607 for rdata in rds:
608 yield (name, rds.ttl, rdata)
610 def to_file(
611 self,
612 f: str | os.PathLike | BinaryIO | TextIO,
613 sorted: bool = True,
614 relativize: bool = True,
615 nl: str | bytes | None = None,
616 want_comments: bool = False,
617 want_origin: bool = False,
618 style: ZoneStyle | None = None,
619 ) -> None:
620 """Write a zone to a file.
622 :param f: A file object, a ``str`` filename, or an ``os.PathLike``.
623 If not a file object, it is treated as the name of a file to
624 open.
625 :param bool sorted: If ``True`` (the default), the file will be
626 written with the names sorted in DNSSEC order from least to
627 greatest. Otherwise the names will be written in whatever order
628 they happen to have in the zone's dictionary.
629 :param bool relativize: If ``True`` (the default), domain names in
630 the output will be relativized to the zone's origin if possible.
631 :param nl: The end of line string. If ``None`` (the default), the
632 output will use the platform's native end-of-line marker (i.e. LF
633 on POSIX, CRLF on Windows).
634 :type nl: str, bytes, or ``None``
635 :param bool want_comments: If ``True``, emit end-of-line comments as
636 part of writing the file. If ``False`` (the default), do not
637 emit them.
638 :param bool want_origin: If ``True``, emit a ``$ORIGIN`` line at the
639 start of the file. If ``False`` (the default), do not emit one.
640 :param style: If specified, the style overrides the other parameters.
641 :type style: :py:class:`dns.zone.ZoneStyle` or ``None``
642 """
643 if style is None:
644 kw = {}
645 kw["sorted"] = sorted
646 kw["relativize"] = relativize
647 if relativize:
648 assert self.origin is not None
649 kw["origin"] = self.origin
650 kw["nl"] = nl
651 kw["want_comments"] = want_comments
652 kw["want_origin"] = want_origin
653 style = ZoneStyle.from_keywords(kw)
654 return self.to_styled_file(style, f)
656 def _write_line(self, output, l, l_b, nl, nl_b):
657 try:
658 bout = cast(BinaryIO, output)
659 bout.write(l_b)
660 bout.write(nl_b)
661 except TypeError: # textual mode
662 tout = cast(TextIO, output)
663 tout.write(l)
664 tout.write(nl)
666 def to_styled_file(
667 self,
668 style: ZoneStyle,
669 f: str | os.PathLike | BinaryIO | TextIO,
670 ) -> None:
671 """Write a zone to a styled file.
673 :param style: The style to apply.
674 :type style: :py:class:`dns.zone.ZoneStyle`
675 :param f: A file object, a ``str`` filename, or an ``os.PathLike``.
676 If not a file object, it is treated as the name of a file to
677 open.
678 """
680 # Apply style items we learned from $UNICODE when we loaded the zone (if any).
681 if style.want_unicode_directive:
682 idna_codec: dns.name.IDNACodec | None = style.idna_codec
683 if "2008" in self.unicode:
684 idna_codec = dns.name.IDNA_2008_Practical
685 elif "2003" in self.unicode:
686 idna_codec = dns.name.IDNA_2003_Practical
687 if "TXT" in self.unicode:
688 txt_is_utf8 = True
689 else:
690 txt_is_utf8 = style.txt_is_utf8
691 style = style.replace(idna_codec=idna_codec, txt_is_utf8=txt_is_utf8)
692 with dns._file_util.maybe_open(f, "wb") as output:
693 # must be in this way, f.encoding may contain None, or even
694 # attribute may not be there
695 file_enc = getattr(f, "encoding", None)
696 if file_enc is None:
697 file_enc = "utf-8"
699 if style.nl is None:
700 # binary mode, '\n' is not enough
701 nl_b = os.linesep.encode(file_enc)
702 nl = "\n"
703 elif isinstance(style.nl, str):
704 nl_b = style.nl.encode(file_enc)
705 nl = style.nl
706 else:
707 nl_b = style.nl
708 nl = style.nl.decode()
709 assert nl is not None
710 assert nl_b is not None
712 if style.want_unicode_directive and len(self.unicode) > 0:
713 l = "$UNICODE " + " ".join(sorted(self.unicode))
714 l_b = l.encode(file_enc)
715 self._write_line(output, l, l_b, nl, nl_b)
716 if style.want_origin:
717 assert self.origin is not None
718 # Ensure we don't relativize the origin to the origin in $ORIGIN!
719 origin_style = style.replace(origin=None)
720 l = "$ORIGIN " + self.origin.to_styled_text(origin_style)
721 l_b = l.encode(file_enc)
722 self._write_line(output, l, l_b, nl, nl_b)
723 if style.default_ttl is not None:
724 l = f"$TTL {style.default_ttl}"
725 l_b = l.encode(file_enc)
726 self._write_line(output, l, l_b, nl, nl_b)
728 if style.sorted:
729 names = list(self.keys())
730 names.sort()
731 else:
732 names = self.keys()
733 for n in names:
734 l = self[n].to_styled_text(style, n)
735 l_b = l.encode(file_enc)
736 self._write_line(output, l, l_b, nl, nl_b)
738 def to_text(
739 self,
740 sorted: bool = True,
741 relativize: bool = True,
742 nl: str | None = None,
743 want_comments: bool = False,
744 want_origin: bool = False,
745 style: ZoneStyle | None = None,
746 ) -> str:
747 """Return a zone's text as though it were written to a file.
749 :param bool sorted: If ``True`` (the default), the output will be
750 written with the names sorted in DNSSEC order from least to
751 greatest. Otherwise the names will be written in whatever order
752 they happen to have in the zone's dictionary.
753 :param bool relativize: If ``True`` (the default), domain names in
754 the output will be relativized to the zone's origin if possible.
755 :param nl: The end of line string. If ``None`` (the default), the
756 output will use the platform's native end-of-line marker (i.e. LF
757 on POSIX, CRLF on Windows).
758 :type nl: str or ``None``
759 :param bool want_comments: If ``True``, emit end-of-line comments as
760 part of the output. If ``False`` (the default), do not emit them.
761 :param bool want_origin: If ``True``, emit a ``$ORIGIN`` line at the
762 start of the output. If ``False`` (the default), do not emit one.
763 :param style: If specified, the style overrides the other parameters.
764 :type style: :py:class:`dns.zone.ZoneStyle` or ``None``
765 :rtype: str
766 """
767 temp_buffer = io.StringIO()
768 self.to_file(
769 temp_buffer, sorted, relativize, nl, want_comments, want_origin, style
770 )
771 return_value = temp_buffer.getvalue()
772 temp_buffer.close()
773 return return_value
775 def to_styled_text(self, style: ZoneStyle):
776 """Return a zone's styled text as though it were written to a file.
778 See the documentation for :py:class:`dns.zone.ZoneStyle` for a description
779 of the style parameters.
781 :param style: The style to apply.
782 :type style: :py:class:`dns.zone.ZoneStyle`
783 :rtype: str
784 """
786 temp_buffer = io.StringIO()
787 self.to_styled_file(style, temp_buffer)
788 return_value = temp_buffer.getvalue()
789 temp_buffer.close()
790 return return_value
792 def check_origin(self) -> None:
793 """Do some simple checking of the zone's origin.
795 :raises dns.zone.NoSOA: if there is no SOA RRset.
796 :raises dns.zone.NoNS: if there is no NS RRset.
797 :raises KeyError: if there is no origin node.
798 """
799 if self.relativize:
800 name = dns.name.empty
801 else:
802 assert self.origin is not None
803 name = self.origin
804 if self.get_rdataset(name, dns.rdatatype.SOA) is None:
805 raise NoSOA
806 if self.get_rdataset(name, dns.rdatatype.NS) is None:
807 raise NoNS
809 def get_soa(
810 self, txn: dns.transaction.Transaction | None = None
811 ) -> dns.rdtypes.ANY.SOA.SOA:
812 """Get the zone SOA rdata.
814 :raises dns.zone.NoSOA: if there is no SOA RRset.
815 :rtype: :py:class:`dns.rdtypes.ANY.SOA.SOA`
816 """
817 if self.relativize:
818 origin_name = dns.name.empty
819 else:
820 if self.origin is None:
821 # get_soa() has been called very early, and there must not be
822 # an SOA if there is no origin.
823 raise NoSOA
824 origin_name = self.origin
825 soa_rds: dns.rdataset.Rdataset | None
826 if txn:
827 soa_rds = txn.get(origin_name, dns.rdatatype.SOA)
828 else:
829 soa_rds = self.get_rdataset(origin_name, dns.rdatatype.SOA)
830 if soa_rds is None:
831 raise NoSOA
832 else:
833 soa = cast(dns.rdtypes.ANY.SOA.SOA, soa_rds[0])
834 return soa
836 def _compute_digest(
837 self,
838 hash_algorithm: DigestHashAlgorithm,
839 scheme: DigestScheme = DigestScheme.SIMPLE,
840 ) -> bytes:
841 hashinfo = _digest_hashers.get(hash_algorithm)
842 if not hashinfo:
843 raise UnsupportedDigestHashAlgorithm
844 if scheme != DigestScheme.SIMPLE:
845 raise UnsupportedDigestScheme
847 if self.relativize:
848 origin_name = dns.name.empty
849 else:
850 assert self.origin is not None
851 origin_name = self.origin
852 hasher = hashinfo()
853 for name, node in sorted(self.items()):
854 rrnamebuf = name.to_digestable(self.origin)
855 for rdataset in sorted(node, key=lambda rds: (rds.rdtype, rds.covers)):
856 if name == origin_name and dns.rdatatype.ZONEMD in (
857 rdataset.rdtype,
858 rdataset.covers,
859 ):
860 continue
861 rrfixed = struct.pack(
862 "!HHI", rdataset.rdtype, rdataset.rdclass, rdataset.ttl
863 )
864 rdatas = [rdata.to_digestable(self.origin) for rdata in rdataset]
865 for rdata in sorted(rdatas):
866 rrlen = struct.pack("!H", len(rdata))
867 hasher.update(rrnamebuf + rrfixed + rrlen + rdata)
868 return hasher.digest()
870 def compute_digest(
871 self,
872 hash_algorithm: DigestHashAlgorithm,
873 scheme: DigestScheme = DigestScheme.SIMPLE,
874 ) -> dns.rdtypes.ANY.ZONEMD.ZONEMD:
875 serial = self.get_soa().serial
876 digest = self._compute_digest(hash_algorithm, scheme)
877 return dns.rdtypes.ANY.ZONEMD.ZONEMD(
878 self.rdclass, dns.rdatatype.ZONEMD, serial, scheme, hash_algorithm, digest
879 )
881 def verify_digest(
882 self, zonemd: dns.rdtypes.ANY.ZONEMD.ZONEMD | None = None
883 ) -> None:
884 digests: dns.rdataset.Rdataset | list[dns.rdtypes.ANY.ZONEMD.ZONEMD]
885 if zonemd:
886 digests = [zonemd]
887 else:
888 assert self.origin is not None
889 rds = self.get_rdataset(self.origin, dns.rdatatype.ZONEMD)
890 if rds is None:
891 raise NoDigest
892 digests = rds
893 try:
894 serial = self.get_soa().serial
895 except NoSOA:
896 # Without an SOA there is no serial for the ZONEMD serial to match.
897 raise DigestVerificationFailure
898 # RFC 8976 section 4 step 4: a (scheme, hash algorithm) tuple that occurs
899 # more than once disqualifies every ZONEMD RR that has it.
900 duplicated: set[tuple[int, int]] = set()
901 seen: set[tuple[int, int]] = set()
902 for digest in digests:
903 key = (digest.scheme, digest.hash_algorithm)
904 if key in seen:
905 duplicated.add(key)
906 seen.add(key)
907 for digest in digests:
908 if (digest.scheme, digest.hash_algorithm) in duplicated:
909 continue
910 # RFC 8976 section 4 step 5.1. The apex ZONEMD RRset is not part of
911 # what the digest covers, so its serial is not tied to the zone's by
912 # the digest comparison below.
913 if digest.serial != serial:
914 continue
915 try:
916 computed = self._compute_digest(digest.hash_algorithm, digest.scheme)
917 if computed == digest.digest:
918 return
919 except Exception:
920 pass
921 raise DigestVerificationFailure
923 # TransactionManager methods
925 def reader(self) -> "Transaction":
926 return Transaction(self, False, Version(self, 1, self.nodes, self.origin))
928 def writer(self, replacement: bool = False) -> "Transaction":
929 txn = Transaction(self, replacement)
930 txn._setup_version()
931 return txn
933 def origin_information(
934 self,
935 ) -> tuple[dns.name.Name | None, bool, dns.name.Name | None]:
936 effective: dns.name.Name | None
937 if self.relativize:
938 effective = dns.name.empty
939 else:
940 effective = self.origin
941 return (self.origin, self.relativize, effective)
943 def get_class(self):
944 return self.rdclass
946 # Transaction methods
948 def _end_read(self, txn):
949 pass
951 def _end_write(self, txn):
952 pass
954 def _commit_version(self, txn, version, origin):
955 self.nodes = version.nodes
956 if self.origin is None:
957 self.origin = origin
959 def _get_next_version_id(self) -> int:
960 # Versions are ephemeral and all have id 1
961 return 1
964# These classes used to be in dns.versioned, but have moved here so we can use
965# the copy-on-write transaction mechanism for both kinds of zones. In a
966# regular zone, the version only exists during the transaction, and the nodes
967# are regular dns.node.Nodes.
969# A node with a version id.
972class VersionedNode(dns.node.Node): # lgtm[py/missing-equals]
973 __slots__ = ["id"]
975 def __init__(self):
976 super().__init__()
977 # A proper id will get set by the Version
978 self.id = 0
981@dns.immutable.immutable
982class ImmutableVersionedNode(VersionedNode):
983 def __init__(self, node):
984 super().__init__()
985 self.id = node.id
986 self.rdatasets = tuple(
987 [dns.rdataset.ImmutableRdataset(rds) for rds in node.rdatasets]
988 )
990 def find_rdataset(
991 self,
992 rdclass: dns.rdataclass.RdataClass,
993 rdtype: dns.rdatatype.RdataType,
994 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
995 create: bool = False,
996 ) -> dns.rdataset.Rdataset:
997 if create:
998 raise TypeError("immutable")
999 return super().find_rdataset(rdclass, rdtype, covers, False)
1001 def get_rdataset(
1002 self,
1003 rdclass: dns.rdataclass.RdataClass,
1004 rdtype: dns.rdatatype.RdataType,
1005 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
1006 create: bool = False,
1007 ) -> dns.rdataset.Rdataset | None:
1008 if create:
1009 raise TypeError("immutable")
1010 return super().get_rdataset(rdclass, rdtype, covers, False)
1012 def delete_rdataset(
1013 self,
1014 rdclass: dns.rdataclass.RdataClass,
1015 rdtype: dns.rdatatype.RdataType,
1016 covers: dns.rdatatype.RdataType = dns.rdatatype.NONE,
1017 ) -> None:
1018 raise TypeError("immutable")
1020 def replace_rdataset(self, replacement: dns.rdataset.Rdataset) -> None:
1021 raise TypeError("immutable")
1023 def is_immutable(self) -> bool:
1024 return True
1027class Version:
1028 def __init__(
1029 self,
1030 zone: Zone,
1031 id: int,
1032 nodes: MutableMapping[dns.name.Name, dns.node.Node] | None = None,
1033 origin: dns.name.Name | None = None,
1034 ):
1035 self.zone = zone
1036 self.id = id
1037 if nodes is not None:
1038 self.nodes = nodes
1039 else:
1040 self.nodes = zone.map_factory()
1041 self.origin = origin
1043 def _validate_name(self, name: dns.name.Name) -> dns.name.Name:
1044 return _validate_name(name, self.origin, self.zone.relativize)
1046 def get_node(self, name: dns.name.Name) -> dns.node.Node | None:
1047 name = self._validate_name(name)
1048 return self.nodes.get(name)
1050 def get_rdataset(
1051 self,
1052 name: dns.name.Name,
1053 rdtype: dns.rdatatype.RdataType,
1054 covers: dns.rdatatype.RdataType,
1055 ) -> dns.rdataset.Rdataset | None:
1056 node = self.get_node(name)
1057 if node is None:
1058 return None
1059 return node.get_rdataset(self.zone.rdclass, rdtype, covers)
1061 def keys(self):
1062 return self.nodes.keys()
1064 def items(self):
1065 return self.nodes.items()
1068class WritableVersion(Version):
1069 def __init__(self, zone: Zone, replacement: bool = False):
1070 # The zone._versions_lock must be held by our caller in a versioned
1071 # zone.
1072 id = zone._get_next_version_id()
1073 super().__init__(zone, id)
1074 if not replacement:
1075 # We copy the map, because that gives us a simple and thread-safe
1076 # way of doing versions, and we have a garbage collector to help
1077 # us. We only make new node objects if we actually change the
1078 # node.
1079 self.nodes.update(zone.nodes)
1080 # We have to copy the zone origin as it may be None in the first
1081 # version, and we don't want to mutate the zone until we commit.
1082 self.origin = zone.origin
1083 self.changed: set[dns.name.Name] = set()
1085 def _maybe_cow_with_name(
1086 self, name: dns.name.Name
1087 ) -> tuple[dns.node.Node, dns.name.Name]:
1088 name = self._validate_name(name)
1089 node = self.nodes.get(name)
1090 if node is None or name not in self.changed:
1091 new_node = self.zone.node_factory()
1092 if hasattr(new_node, "id"):
1093 # We keep doing this for backwards compatibility, as earlier
1094 # code used new_node.id != self.id for the "do we need to CoW?"
1095 # test. Now we use the changed set as this works with both
1096 # regular zones and versioned zones.
1097 #
1098 # We ignore the type error as this is safe but checkers have trouble.
1099 new_node.id = self.id # type: ignore
1100 if node is not None:
1101 # moo! copy on write!
1102 new_node.rdatasets.extend(node.rdatasets)
1103 self.nodes[name] = new_node
1104 self.changed.add(name)
1105 return (new_node, name)
1106 else:
1107 return (node, name)
1109 def _maybe_cow(self, name: dns.name.Name) -> dns.node.Node:
1110 return self._maybe_cow_with_name(name)[0]
1112 def delete_node(self, name: dns.name.Name) -> None:
1113 name = self._validate_name(name)
1114 if name in self.nodes:
1115 del self.nodes[name]
1116 self.changed.add(name)
1118 def put_rdataset(
1119 self, name: dns.name.Name, rdataset: dns.rdataset.Rdataset
1120 ) -> None:
1121 node = self._maybe_cow(name)
1122 node.replace_rdataset(rdataset)
1124 def delete_rdataset(
1125 self,
1126 name: dns.name.Name,
1127 rdtype: dns.rdatatype.RdataType,
1128 covers: dns.rdatatype.RdataType,
1129 ) -> None:
1130 node = self._maybe_cow(name)
1131 node.delete_rdataset(self.zone.rdclass, rdtype, covers)
1132 if len(node) == 0:
1133 del self.nodes[name]
1136@dns.immutable.immutable
1137class ImmutableVersion(Version):
1138 def __init__(self, version: Version):
1139 if not isinstance(version, WritableVersion):
1140 raise ValueError(
1141 "a dns.zone.ImmutableVersion requires a dns.zone.WritableVersion"
1142 )
1143 # We tell super() that it's a replacement as we don't want it
1144 # to copy the nodes, as we're about to do that with an
1145 # immutable Dict.
1146 super().__init__(version.zone, True)
1147 # set the right id!
1148 self.id = version.id
1149 # keep the origin
1150 self.origin = version.origin
1151 # Make changed nodes immutable
1152 for name in version.changed:
1153 node = version.nodes.get(name)
1154 # it might not exist if we deleted it in the version
1155 if node:
1156 version.nodes[name] = ImmutableVersionedNode(node)
1157 # We're changing the type of the nodes dictionary here on purpose, so
1158 # we ignore the mypy error.
1159 self.nodes = dns.immutable.Dict(
1160 version.nodes, True, self.zone.map_factory
1161 ) # pyright: ignore
1164class Transaction(dns.transaction.Transaction):
1165 def __init__(
1166 self,
1167 zone: Zone,
1168 replacement: bool,
1169 version: Version | None = None,
1170 make_immutable: bool = False,
1171 ):
1172 read_only = version is not None
1173 super().__init__(zone, replacement, read_only)
1174 self.version = version
1175 self.make_immutable = make_immutable
1177 @property
1178 def zone(self) -> Zone:
1179 return cast(Zone, self.manager)
1181 def _setup_version(self):
1182 assert self.version is None
1183 factory = self.manager.writable_version_factory # type: ignore
1184 if factory is None:
1185 factory = WritableVersion
1186 self.version = factory(self.zone, self.replacement) # pyright: ignore
1188 def _get_rdataset(self, name, rdtype, covers):
1189 assert self.version is not None
1190 return self.version.get_rdataset(name, rdtype, covers)
1192 def _put_rdataset(self, name, rdataset):
1193 assert not self.read_only
1194 assert self.version is not None
1195 version = cast(WritableVersion, self.version)
1196 version.put_rdataset(name, rdataset)
1198 def _delete_name(self, name):
1199 assert not self.read_only
1200 assert self.version is not None
1201 version = cast(WritableVersion, self.version)
1202 version.delete_node(name)
1204 def _delete_rdataset(self, name, rdtype, covers):
1205 assert not self.read_only
1206 assert self.version is not None
1207 version = cast(WritableVersion, self.version)
1208 version.delete_rdataset(name, rdtype, covers)
1210 def _name_exists(self, name):
1211 assert self.version is not None
1212 return self.version.get_node(name) is not None
1214 def _changed(self):
1215 if self.read_only:
1216 return False
1217 else:
1218 assert self.version is not None
1219 version = cast(WritableVersion, self.version)
1220 return len(version.changed) > 0
1222 def _end_transaction(self, commit):
1223 assert self.zone is not None
1224 assert self.version is not None
1225 if self.read_only:
1226 self.zone._end_read(self) # pyright: ignore
1227 elif commit and len(cast(WritableVersion, self.version).changed) > 0:
1228 if self.make_immutable:
1229 factory = self.manager.immutable_version_factory # type: ignore
1230 if factory is None:
1231 factory = ImmutableVersion
1232 version = factory(self.version)
1233 else:
1234 version = self.version
1235 self.zone._commit_version( # pyright: ignore
1236 self, version, self.version.origin
1237 )
1239 else:
1240 # rollback
1241 self.zone._end_write(self) # pyright: ignore
1243 def _set_origin(self, origin):
1244 assert self.version is not None
1245 if self.version.origin is None:
1246 self.version.origin = origin
1248 def _iterate_rdatasets(self):
1249 assert self.version is not None
1250 for name, node in self.version.items():
1251 for rdataset in node:
1252 yield (name, rdataset)
1254 def _iterate_names(self):
1255 assert self.version is not None
1256 return self.version.keys()
1258 def _get_node(self, name):
1259 assert self.version is not None
1260 return self.version.get_node(name)
1262 def _origin_information(
1263 self,
1264 ) -> tuple[dns.name.Name | None, bool, dns.name.Name | None]:
1265 assert self.version is not None
1266 absolute, relativize, effective = self.manager.origin_information()
1267 if absolute is None and self.version.origin is not None:
1268 # No origin has been committed yet, but we've learned one as part of
1269 # this txn. Use it.
1270 absolute = self.version.origin
1271 if relativize:
1272 effective = dns.name.empty
1273 else:
1274 effective = absolute
1275 return (absolute, relativize, effective)
1278def _from_text(
1279 text: Any,
1280 origin: dns.name.Name | str | None = None,
1281 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
1282 relativize: bool = True,
1283 zone_factory: Any = Zone,
1284 filename: str | None = None,
1285 allow_include: bool = False,
1286 check_origin: bool = True,
1287 idna_codec: dns.name.IDNACodec | None = None,
1288 allow_directives: bool | Iterable[str] = True,
1289 transaction_setup: Callable[[dns.transaction.Transaction], None] | None = None,
1290 rfc2308_ttl: bool = True,
1291) -> Zone:
1292 # See the comments for the public APIs from_text() and from_file() for
1293 # details.
1295 # 'text' can also be a file, but we don't publish that fact
1296 # since it's an implementation detail. The official file
1297 # interface is from_file().
1299 if filename is None:
1300 filename = "<string>"
1301 zone = zone_factory(origin, rdclass, relativize=relativize)
1302 with zone.writer(True) as txn:
1303 if transaction_setup is not None:
1304 transaction_setup(txn)
1305 tok = dns.tokenizer.Tokenizer(text, filename, idna_codec=idna_codec)
1306 reader = dns.zonefile.Reader(
1307 tok,
1308 rdclass,
1309 txn,
1310 allow_include=allow_include,
1311 allow_directives=allow_directives,
1312 rfc2308_ttl=rfc2308_ttl,
1313 )
1314 try:
1315 reader.read()
1316 zone.unicode = txn.unicode
1317 except dns.zonefile.UnknownOrigin:
1318 # for backwards compatibility
1319 raise UnknownOrigin
1320 # Now that we're done reading, do some basic checking of the zone.
1321 if check_origin:
1322 zone.check_origin()
1323 return zone
1326def from_text(
1327 text: str,
1328 origin: dns.name.Name | str | None = None,
1329 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
1330 relativize: bool = True,
1331 zone_factory: Any = Zone,
1332 filename: str | None = None,
1333 allow_include: bool = False,
1334 check_origin: bool = True,
1335 idna_codec: dns.name.IDNACodec | None = None,
1336 allow_directives: bool | Iterable[str] = True,
1337 transaction_setup: Callable[[dns.transaction.Transaction], None] | None = None,
1338 rfc2308_ttl: bool = True,
1339) -> Zone:
1340 """Build a zone object from a zone file format string.
1342 :param str text: The zone file format input.
1343 :param origin: The origin of the zone. If not specified, the first
1344 ``$ORIGIN`` statement in the zone file will determine the origin.
1345 :type origin: :py:class:`dns.name.Name`, str, or ``None``
1346 :param rdclass: The zone's rdata class; the default is class IN.
1347 :type rdclass: :py:class:`dns.rdataclass.RdataClass`
1348 :param bool relativize: Whether domain names are relativized to the
1349 zone's origin. The default is ``True``.
1350 :param zone_factory: The zone factory to use. If ``None``, then
1351 :py:class:`dns.zone.Zone` will be used. The value may be any class
1352 or callable that returns a subclass of :py:class:`dns.zone.Zone`.
1353 :param filename: The filename to emit when describing where an error
1354 occurred; the default is ``'<string>'``.
1355 :type filename: str or ``None``
1356 :param bool allow_include: If ``True`` (the default), then ``$INCLUDE``
1357 directives are permitted. If ``False``, then encountering a
1358 ``$INCLUDE`` will raise a :py:exc:`SyntaxError`.
1359 :param bool check_origin: If ``True`` (the default), then sanity checks
1360 of the origin node will be made by calling
1361 :py:meth:`dns.zone.Zone.check_origin`.
1362 :param idna_codec: The IDNA encoder/decoder. If ``None``, the default
1363 IDNA encoder/decoder is used.
1364 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
1365 :param allow_directives: If ``True`` (the default), directives are
1366 permitted and *allow_include* controls ``$INCLUDE``. If ``False``
1367 or an empty iterable, no directive processing is done. If a
1368 non-empty iterable, only the listed directives (including the ``$``)
1369 are allowed.
1370 :type allow_directives: bool or Iterable[str]
1371 :param transaction_setup: If not ``None``, it is called with the just-created
1372 transaction. This lets the caller alter the transaction's configuration
1373 before it is used, for example adding checking policies.
1374 :type transaction_setup: ``None`` or ``Callable[[dns.transaction.Transaction], None]``.
1375 :param bool rfc2308_ttl: If ``True``, the default, then RFC 2308 section 4
1376 is followed and the SOA MINIMUM field is not treated as a source of the
1377 default TTL; a record with no TTL instead inherits the most recently
1378 stated TTL, per RFC 1035 section 5.1. The SOA MINIMUM is still used as a
1379 last resort if no TTL has been stated at all. If ``False``, then a master
1380 file with no ``$TTL`` directive gets its default TTL from the SOA MINIMUM
1381 field, which is the pre-RFC 2308 behavior.
1382 :raises dns.zone.NoSOA: if there is no SOA RRset.
1383 :raises dns.zone.NoNS: if there is no NS RRset.
1384 :raises KeyError: if there is no origin node.
1385 :returns: A subclass of :py:class:`dns.zone.Zone`.
1386 """
1387 return _from_text(
1388 text,
1389 origin,
1390 rdclass,
1391 relativize,
1392 zone_factory,
1393 filename,
1394 allow_include,
1395 check_origin,
1396 idna_codec,
1397 allow_directives,
1398 transaction_setup,
1399 rfc2308_ttl,
1400 )
1403def from_file(
1404 f: str | os.PathLike | TextIO,
1405 origin: dns.name.Name | str | None = None,
1406 rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
1407 relativize: bool = True,
1408 zone_factory: Any = Zone,
1409 filename: str | None = None,
1410 allow_include: bool = True,
1411 check_origin: bool = True,
1412 idna_codec: dns.name.IDNACodec | None = None,
1413 allow_directives: bool | Iterable[str] = True,
1414 transaction_setup: Callable[[dns.transaction.Transaction], None] | None = None,
1415 rfc2308_ttl: bool = True,
1416) -> Zone:
1417 """Read a zone file and build a zone object.
1419 :param f: A file object, a ``str`` filename, or an ``os.PathLike``. If
1420 not a file object, it is treated as the name of a file to open.
1421 :param origin: The origin of the zone. If not specified, the first
1422 ``$ORIGIN`` statement in the zone file will determine the origin.
1423 :type origin: :py:class:`dns.name.Name`, str, or ``None``
1424 :param rdclass: The zone's rdata class; the default is class IN.
1425 :type rdclass: :py:class:`dns.rdataclass.RdataClass`
1426 :param bool relativize: Whether domain names are relativized to the
1427 zone's origin. The default is ``True``.
1428 :param zone_factory: The zone factory to use. If ``None``, then
1429 :py:class:`dns.zone.Zone` will be used. The value may be any class
1430 or callable that returns a subclass of :py:class:`dns.zone.Zone`.
1431 :param filename: The filename to emit when describing where an error
1432 occurred; the default is ``'<string>'``.
1433 :type filename: str or ``None``
1434 :param bool allow_include: If ``True`` (the default), then ``$INCLUDE``
1435 directives are permitted. If ``False``, then encountering a
1436 ``$INCLUDE`` will raise a :py:exc:`SyntaxError`.
1437 :param bool check_origin: If ``True`` (the default), then sanity checks
1438 of the origin node will be made by calling
1439 :py:meth:`dns.zone.Zone.check_origin`.
1440 :param idna_codec: The IDNA encoder/decoder. If ``None``, the default
1441 IDNA encoder/decoder is used.
1442 :type idna_codec: :py:class:`dns.name.IDNACodec` or ``None``
1443 :param allow_directives: If ``True`` (the default), directives are
1444 permitted and *allow_include* controls ``$INCLUDE``. If ``False``
1445 or an empty iterable, no directive processing is done. If a
1446 non-empty iterable, only the listed directives (including the ``$``)
1447 are allowed.
1448 :type allow_directives: bool or Iterable[str]
1449 :param transaction_setup: If not ``None``, it is called with the just-created
1450 transaction. This lets the caller alter the transaction's configuration
1451 before it is used, for example adding checking policies.
1452 :type transaction_setup: ``None`` or ``Callable[[dns.transaction.Transaction], None]``.
1453 :param bool rfc2308_ttl: If ``True``, the default, then RFC 2308 section 4
1454 is followed and the SOA MINIMUM field is not treated as a source of the
1455 default TTL; a record with no TTL instead inherits the most recently
1456 stated TTL, per RFC 1035 section 5.1. The SOA MINIMUM is still used as a
1457 last resort if no TTL has been stated at all. If ``False``, then a master
1458 file with no ``$TTL`` directive gets its default TTL from the SOA MINIMUM
1459 field, which is the pre-RFC 2308 behavior.
1460 :raises dns.zone.NoSOA: if there is no SOA RRset.
1461 :raises dns.zone.NoNS: if there is no NS RRset.
1462 :raises KeyError: if there is no origin node.
1463 :returns: A subclass of :py:class:`dns.zone.Zone`.
1464 """
1466 if filename is None:
1467 filename = dns._file_util.as_filename(f)
1468 with dns._file_util.maybe_open(f, encoding="utf-8") as fp:
1469 return _from_text(
1470 fp,
1471 origin,
1472 rdclass,
1473 relativize,
1474 zone_factory,
1475 filename,
1476 allow_include,
1477 check_origin,
1478 idna_codec,
1479 allow_directives,
1480 transaction_setup,
1481 rfc2308_ttl,
1482 )
1483 assert False # make mypy happy lgtm[py/unreachable-statement]
1486def from_xfr(
1487 xfr: Any,
1488 zone_factory: Any = Zone,
1489 relativize: bool = True,
1490 check_origin: bool = True,
1491) -> Zone:
1492 """Convert the output of a zone transfer generator into a zone object.
1494 :param xfr: A generator of :py:class:`dns.message.Message` objects,
1495 typically from :py:func:`dns.query.xfr`.
1496 :param zone_factory: The zone factory to use. If ``None``, then
1497 :py:class:`dns.zone.Zone` will be used.
1498 :param bool relativize: Whether domain names are relativized to the
1499 zone's origin. The default is ``True``. It is essential that this
1500 setting matches the one specified to the generator.
1501 :param bool check_origin: If ``True`` (the default), then sanity checks
1502 of the origin node will be made by calling
1503 :py:meth:`dns.zone.Zone.check_origin`.
1504 :raises dns.zone.NoSOA: if there is no SOA RRset.
1505 :raises dns.zone.NoNS: if there is no NS RRset.
1506 :raises KeyError: if there is no origin node.
1507 :raises ValueError: if no messages are yielded by the generator.
1508 :returns: A subclass of :py:class:`dns.zone.Zone`.
1509 """
1511 z = None
1512 for r in xfr:
1513 if z is None:
1514 if relativize:
1515 origin = r.origin
1516 else:
1517 origin = r.answer[0].name
1518 rdclass = r.answer[0].rdclass
1519 z = zone_factory(origin, rdclass, relativize=relativize)
1520 for rrset in r.answer:
1521 znode = z.nodes.get(rrset.name)
1522 if not znode:
1523 znode = z.node_factory()
1524 z.nodes[rrset.name] = znode
1525 zrds = znode.find_rdataset(rrset.rdclass, rrset.rdtype, rrset.covers, True)
1526 zrds.update_ttl(rrset.ttl)
1527 for rd in rrset:
1528 zrds.add(rd)
1529 if z is None:
1530 raise ValueError("empty transfer")
1531 if check_origin:
1532 z.check_origin()
1533 return z