codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3# Copyright (C) 2003-2007, 2009-2011 Nominum, Inc.4#5# Permission to use, copy, modify, and distribute this software and its6# documentation for any purpose with or without fee is hereby granted,7# provided that the above copyright notice and this permission notice8# appear in all copies.9#10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.17 18"""DNS Zones."""19 20import re21import sys22from typing import Any, Iterable, List, Optional, Set, Tuple, Union23 24import dns.exception25import dns.grange26import dns.name27import dns.node28import dns.rdata29import dns.rdataclass30import dns.rdatatype31import dns.rdtypes.ANY.SOA32import dns.rrset33import dns.tokenizer34import dns.transaction35import dns.ttl36 37 38class UnknownOrigin(dns.exception.DNSException):39    """Unknown origin"""40 41 42class CNAMEAndOtherData(dns.exception.DNSException):43    """A node has a CNAME and other data"""44 45 46def _check_cname_and_other_data(txn, name, rdataset):47    rdataset_kind = dns.node.NodeKind.classify_rdataset(rdataset)48    node = txn.get_node(name)49    if node is None:50        # empty nodes are neutral.51        return52    node_kind = node.classify()53    if (54        node_kind == dns.node.NodeKind.CNAME55        and rdataset_kind == dns.node.NodeKind.REGULAR56    ):57        raise CNAMEAndOtherData("rdataset type is not compatible with a CNAME node")58    elif (59        node_kind == dns.node.NodeKind.REGULAR60        and rdataset_kind == dns.node.NodeKind.CNAME61    ):62        raise CNAMEAndOtherData(63            "CNAME rdataset is not compatible with a regular data node"64        )65    # Otherwise at least one of the node and the rdataset is neutral, so66    # adding the rdataset is ok67 68 69SavedStateType = Tuple[70    dns.tokenizer.Tokenizer,71    Optional[dns.name.Name],  # current_origin72    Optional[dns.name.Name],  # last_name73    Optional[Any],  # current_file74    int,  # last_ttl75    bool,  # last_ttl_known76    int,  # default_ttl77    bool,78]  # default_ttl_known79 80 81def _upper_dollarize(s):82    s = s.upper()83    if not s.startswith("$"):84        s = "$" + s85    return s86 87 88class Reader:89    """Read a DNS zone file into a transaction."""90 91    def __init__(92        self,93        tok: dns.tokenizer.Tokenizer,94        rdclass: dns.rdataclass.RdataClass,95        txn: dns.transaction.Transaction,96        allow_include: bool = False,97        allow_directives: Union[bool, Iterable[str]] = True,98        force_name: Optional[dns.name.Name] = None,99        force_ttl: Optional[int] = None,100        force_rdclass: Optional[dns.rdataclass.RdataClass] = None,101        force_rdtype: Optional[dns.rdatatype.RdataType] = None,102        default_ttl: Optional[int] = None,103    ):104        self.tok = tok105        (self.zone_origin, self.relativize, _) = txn.manager.origin_information()106        self.current_origin = self.zone_origin107        self.last_ttl = 0108        self.last_ttl_known = False109        if force_ttl is not None:110            default_ttl = force_ttl111        if default_ttl is None:112            self.default_ttl = 0113            self.default_ttl_known = False114        else:115            self.default_ttl = default_ttl116            self.default_ttl_known = True117        self.last_name = self.current_origin118        self.zone_rdclass = rdclass119        self.txn = txn120        self.saved_state: List[SavedStateType] = []121        self.current_file: Optional[Any] = None122        self.allowed_directives: Set[str]123        if allow_directives is True:124            self.allowed_directives = {"$GENERATE", "$ORIGIN", "$TTL"}125            if allow_include:126                self.allowed_directives.add("$INCLUDE")127        elif allow_directives is False:128            # allow_include was ignored in earlier releases if allow_directives was129            # False, so we continue that.130            self.allowed_directives = set()131        else:132            # Note that if directives are explicitly specified, then allow_include133            # is ignored.134            self.allowed_directives = set(_upper_dollarize(d) for d in allow_directives)135        self.force_name = force_name136        self.force_ttl = force_ttl137        self.force_rdclass = force_rdclass138        self.force_rdtype = force_rdtype139        self.txn.check_put_rdataset(_check_cname_and_other_data)140 141    def _eat_line(self):142        while 1:143            token = self.tok.get()144            if token.is_eol_or_eof():145                break146 147    def _get_identifier(self):148        token = self.tok.get()149        if not token.is_identifier():150            raise dns.exception.SyntaxError151        return token152 153    def _rr_line(self):154        """Process one line from a DNS zone file."""155        token = None156        # Name157        if self.force_name is not None:158            name = self.force_name159        else:160            if self.current_origin is None:161                raise UnknownOrigin162            token = self.tok.get(want_leading=True)163            if not token.is_whitespace():164                self.last_name = self.tok.as_name(token, self.current_origin)165            else:166                token = self.tok.get()167                if token.is_eol_or_eof():168                    # treat leading WS followed by EOL/EOF as if they were EOL/EOF.169                    return170                self.tok.unget(token)171            name = self.last_name172            if not name.is_subdomain(self.zone_origin):173                self._eat_line()174                return175            if self.relativize:176                name = name.relativize(self.zone_origin)177 178        # TTL179        if self.force_ttl is not None:180            ttl = self.force_ttl181            self.last_ttl = ttl182            self.last_ttl_known = True183        else:184            token = self._get_identifier()185            ttl = None186            try:187                ttl = dns.ttl.from_text(token.value)188                self.last_ttl = ttl189                self.last_ttl_known = True190                token = None191            except dns.ttl.BadTTL:192                self.tok.unget(token)193 194        # Class195        if self.force_rdclass is not None:196            rdclass = self.force_rdclass197        else:198            token = self._get_identifier()199            try:200                rdclass = dns.rdataclass.from_text(token.value)201            except dns.exception.SyntaxError:202                raise203            except Exception:204                rdclass = self.zone_rdclass205                self.tok.unget(token)206            if rdclass != self.zone_rdclass:207                raise dns.exception.SyntaxError("RR class is not zone's class")208 209        if ttl is None:210            # support for <class> <ttl> <type> syntax211            token = self._get_identifier()212            ttl = None213            try:214                ttl = dns.ttl.from_text(token.value)215                self.last_ttl = ttl216                self.last_ttl_known = True217                token = None218            except dns.ttl.BadTTL:219                if self.default_ttl_known:220                    ttl = self.default_ttl221                elif self.last_ttl_known:222                    ttl = self.last_ttl223                self.tok.unget(token)224 225        # Type226        if self.force_rdtype is not None:227            rdtype = self.force_rdtype228        else:229            token = self._get_identifier()230            try:231                rdtype = dns.rdatatype.from_text(token.value)232            except Exception:233                raise dns.exception.SyntaxError("unknown rdatatype '%s'" % token.value)234 235        try:236            rd = dns.rdata.from_text(237                rdclass,238                rdtype,239                self.tok,240                self.current_origin,241                self.relativize,242                self.zone_origin,243            )244        except dns.exception.SyntaxError:245            # Catch and reraise.246            raise247        except Exception:248            # All exceptions that occur in the processing of rdata249            # are treated as syntax errors.  This is not strictly250            # correct, but it is correct almost all of the time.251            # We convert them to syntax errors so that we can emit252            # helpful filename:line info.253            (ty, va) = sys.exc_info()[:2]254            raise dns.exception.SyntaxError(255                "caught exception {}: {}".format(str(ty), str(va))256            )257 258        if not self.default_ttl_known and rdtype == dns.rdatatype.SOA:259            # The pre-RFC2308 and pre-BIND9 behavior inherits the zone default260            # TTL from the SOA minttl if no $TTL statement is present before the261            # SOA is parsed.262            self.default_ttl = rd.minimum263            self.default_ttl_known = True264            if ttl is None:265                # if we didn't have a TTL on the SOA, set it!266                ttl = rd.minimum267 268        # TTL check.  We had to wait until now to do this as the SOA RR's269        # own TTL can be inferred from its minimum.270        if ttl is None:271            raise dns.exception.SyntaxError("Missing default TTL value")272 273        self.txn.add(name, ttl, rd)274 275    def _parse_modify(self, side: str) -> Tuple[str, str, int, int, str]:276        # Here we catch everything in '{' '}' in a group so we can replace it277        # with ''.278        is_generate1 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+),(.)}).*$")279        is_generate2 = re.compile(r"^.*\$({(\+|-?)(\d+)}).*$")280        is_generate3 = re.compile(r"^.*\$({(\+|-?)(\d+),(\d+)}).*$")281        # Sometimes there are modifiers in the hostname. These come after282        # the dollar sign. They are in the form: ${offset[,width[,base]]}.283        # Make names284        g1 = is_generate1.match(side)285        if g1:286            mod, sign, offset, width, base = g1.groups()287            if sign == "":288                sign = "+"289        g2 = is_generate2.match(side)290        if g2:291            mod, sign, offset = g2.groups()292            if sign == "":293                sign = "+"294            width = 0295            base = "d"296        g3 = is_generate3.match(side)297        if g3:298            mod, sign, offset, width = g3.groups()299            if sign == "":300                sign = "+"301            base = "d"302 303        if not (g1 or g2 or g3):304            mod = ""305            sign = "+"306            offset = 0307            width = 0308            base = "d"309 310        offset = int(offset)311        width = int(width)312 313        if sign not in ["+", "-"]:314            raise dns.exception.SyntaxError("invalid offset sign %s" % sign)315        if base not in ["d", "o", "x", "X", "n", "N"]:316            raise dns.exception.SyntaxError("invalid type %s" % base)317 318        return mod, sign, offset, width, base319 320    def _generate_line(self):321        # range lhs [ttl] [class] type rhs [ comment ]322        """Process one line containing the GENERATE statement from a DNS323        zone file."""324        if self.current_origin is None:325            raise UnknownOrigin326 327        token = self.tok.get()328        # Range (required)329        try:330            start, stop, step = dns.grange.from_text(token.value)331            token = self.tok.get()332            if not token.is_identifier():333                raise dns.exception.SyntaxError334        except Exception:335            raise dns.exception.SyntaxError336 337        # lhs (required)338        try:339            lhs = token.value340            token = self.tok.get()341            if not token.is_identifier():342                raise dns.exception.SyntaxError343        except Exception:344            raise dns.exception.SyntaxError345 346        # TTL347        try:348            ttl = dns.ttl.from_text(token.value)349            self.last_ttl = ttl350            self.last_ttl_known = True351            token = self.tok.get()352            if not token.is_identifier():353                raise dns.exception.SyntaxError354        except dns.ttl.BadTTL:355            if not (self.last_ttl_known or self.default_ttl_known):356                raise dns.exception.SyntaxError("Missing default TTL value")357            if self.default_ttl_known:358                ttl = self.default_ttl359            elif self.last_ttl_known:360                ttl = self.last_ttl361        # Class362        try:363            rdclass = dns.rdataclass.from_text(token.value)364            token = self.tok.get()365            if not token.is_identifier():366                raise dns.exception.SyntaxError367        except dns.exception.SyntaxError:368            raise dns.exception.SyntaxError369        except Exception:370            rdclass = self.zone_rdclass371        if rdclass != self.zone_rdclass:372            raise dns.exception.SyntaxError("RR class is not zone's class")373        # Type374        try:375            rdtype = dns.rdatatype.from_text(token.value)376            token = self.tok.get()377            if not token.is_identifier():378                raise dns.exception.SyntaxError379        except Exception:380            raise dns.exception.SyntaxError("unknown rdatatype '%s'" % token.value)381 382        # rhs (required)383        rhs = token.value384 385        def _calculate_index(counter: int, offset_sign: str, offset: int) -> int:386            """Calculate the index from the counter and offset."""387            if offset_sign == "-":388                offset *= -1389            return counter + offset390 391        def _format_index(index: int, base: str, width: int) -> str:392            """Format the index with the given base, and zero-fill it393            to the given width."""394            if base in ["d", "o", "x", "X"]:395                return format(index, base).zfill(width)396 397            # base can only be n or N here398            hexa = _format_index(index, "x", width)399            nibbles = ".".join(hexa[::-1])[:width]400            if base == "N":401                nibbles = nibbles.upper()402            return nibbles403 404        lmod, lsign, loffset, lwidth, lbase = self._parse_modify(lhs)405        rmod, rsign, roffset, rwidth, rbase = self._parse_modify(rhs)406        for i in range(start, stop + 1, step):407            # +1 because bind is inclusive and python is exclusive408 409            lindex = _calculate_index(i, lsign, loffset)410            rindex = _calculate_index(i, rsign, roffset)411 412            lzfindex = _format_index(lindex, lbase, lwidth)413            rzfindex = _format_index(rindex, rbase, rwidth)414 415            name = lhs.replace("$%s" % (lmod), lzfindex)416            rdata = rhs.replace("$%s" % (rmod), rzfindex)417 418            self.last_name = dns.name.from_text(419                name, self.current_origin, self.tok.idna_codec420            )421            name = self.last_name422            if not name.is_subdomain(self.zone_origin):423                self._eat_line()424                return425            if self.relativize:426                name = name.relativize(self.zone_origin)427 428            try:429                rd = dns.rdata.from_text(430                    rdclass,431                    rdtype,432                    rdata,433                    self.current_origin,434                    self.relativize,435                    self.zone_origin,436                )437            except dns.exception.SyntaxError:438                # Catch and reraise.439                raise440            except Exception:441                # All exceptions that occur in the processing of rdata442                # are treated as syntax errors.  This is not strictly443                # correct, but it is correct almost all of the time.444                # We convert them to syntax errors so that we can emit445                # helpful filename:line info.446                (ty, va) = sys.exc_info()[:2]447                raise dns.exception.SyntaxError(448                    "caught exception %s: %s" % (str(ty), str(va))449                )450 451            self.txn.add(name, ttl, rd)452 453    def read(self) -> None:454        """Read a DNS zone file and build a zone object.455 456        @raises dns.zone.NoSOA: No SOA RR was found at the zone origin457        @raises dns.zone.NoNS: No NS RRset was found at the zone origin458        """459 460        try:461            while 1:462                token = self.tok.get(True, True)463                if token.is_eof():464                    if self.current_file is not None:465                        self.current_file.close()466                    if len(self.saved_state) > 0:467                        (468                            self.tok,469                            self.current_origin,470                            self.last_name,471                            self.current_file,472                            self.last_ttl,473                            self.last_ttl_known,474                            self.default_ttl,475                            self.default_ttl_known,476                        ) = self.saved_state.pop(-1)477                        continue478                    break479                elif token.is_eol():480                    continue481                elif token.is_comment():482                    self.tok.get_eol()483                    continue484                elif token.value[0] == "$" and len(self.allowed_directives) > 0:485                    # Note that we only run directive processing code if at least486                    # one directive is allowed in order to be backwards compatible487                    c = token.value.upper()488                    if c not in self.allowed_directives:489                        raise dns.exception.SyntaxError(490                            f"zone file directive '{c}' is not allowed"491                        )492                    if c == "$TTL":493                        token = self.tok.get()494                        if not token.is_identifier():495                            raise dns.exception.SyntaxError("bad $TTL")496                        self.default_ttl = dns.ttl.from_text(token.value)497                        self.default_ttl_known = True498                        self.tok.get_eol()499                    elif c == "$ORIGIN":500                        self.current_origin = self.tok.get_name()501                        self.tok.get_eol()502                        if self.zone_origin is None:503                            self.zone_origin = self.current_origin504                        self.txn._set_origin(self.current_origin)505                    elif c == "$INCLUDE":506                        token = self.tok.get()507                        filename = token.value508                        token = self.tok.get()509                        new_origin: Optional[dns.name.Name]510                        if token.is_identifier():511                            new_origin = dns.name.from_text(512                                token.value, self.current_origin, self.tok.idna_codec513                            )514                            self.tok.get_eol()515                        elif not token.is_eol_or_eof():516                            raise dns.exception.SyntaxError("bad origin in $INCLUDE")517                        else:518                            new_origin = self.current_origin519                        self.saved_state.append(520                            (521                                self.tok,522                                self.current_origin,523                                self.last_name,524                                self.current_file,525                                self.last_ttl,526                                self.last_ttl_known,527                                self.default_ttl,528                                self.default_ttl_known,529                            )530                        )531                        self.current_file = open(filename, "r")532                        self.tok = dns.tokenizer.Tokenizer(self.current_file, filename)533                        self.current_origin = new_origin534                    elif c == "$GENERATE":535                        self._generate_line()536                    else:537                        raise dns.exception.SyntaxError(538                            f"Unknown zone file directive '{c}'"539                        )540                    continue541                self.tok.unget(token)542                self._rr_line()543        except dns.exception.SyntaxError as detail:544            (filename, line_number) = self.tok.where()545            if detail is None:546                detail = "syntax error"547            ex = dns.exception.SyntaxError(548                "%s:%d: %s" % (filename, line_number, detail)549            )550            tb = sys.exc_info()[2]551            raise ex.with_traceback(tb) from None552 553 554class RRsetsReaderTransaction(dns.transaction.Transaction):555    def __init__(self, manager, replacement, read_only):556        assert not read_only557        super().__init__(manager, replacement, read_only)558        self.rdatasets = {}559 560    def _get_rdataset(self, name, rdtype, covers):561        return self.rdatasets.get((name, rdtype, covers))562 563    def _get_node(self, name):564        rdatasets = []565        for (rdataset_name, _, _), rdataset in self.rdatasets.items():566            if name == rdataset_name:567                rdatasets.append(rdataset)568        if len(rdatasets) == 0:569            return None570        node = dns.node.Node()571        node.rdatasets = rdatasets572        return node573 574    def _put_rdataset(self, name, rdataset):575        self.rdatasets[(name, rdataset.rdtype, rdataset.covers)] = rdataset576 577    def _delete_name(self, name):578        # First remove any changes involving the name579        remove = []580        for key in self.rdatasets:581            if key[0] == name:582                remove.append(key)583        if len(remove) > 0:584            for key in remove:585                del self.rdatasets[key]586 587    def _delete_rdataset(self, name, rdtype, covers):588        try:589            del self.rdatasets[(name, rdtype, covers)]590        except KeyError:591            pass592 593    def _name_exists(self, name):594        for n, _, _ in self.rdatasets:595            if n == name:596                return True597        return False598 599    def _changed(self):600        return len(self.rdatasets) > 0601 602    def _end_transaction(self, commit):603        if commit and self._changed():604            rrsets = []605            for (name, _, _), rdataset in self.rdatasets.items():606                rrset = dns.rrset.RRset(607                    name, rdataset.rdclass, rdataset.rdtype, rdataset.covers608                )609                rrset.update(rdataset)610                rrsets.append(rrset)611            self.manager.set_rrsets(rrsets)612 613    def _set_origin(self, origin):614        pass615 616    def _iterate_rdatasets(self):617        raise NotImplementedError  # pragma: no cover618 619    def _iterate_names(self):620        raise NotImplementedError  # pragma: no cover621 622 623class RRSetsReaderManager(dns.transaction.TransactionManager):624    def __init__(625        self, origin=dns.name.root, relativize=False, rdclass=dns.rdataclass.IN626    ):627        self.origin = origin628        self.relativize = relativize629        self.rdclass = rdclass630        self.rrsets = []631 632    def reader(self):  # pragma: no cover633        raise NotImplementedError634 635    def writer(self, replacement=False):636        assert replacement is True637        return RRsetsReaderTransaction(self, True, False)638 639    def get_class(self):640        return self.rdclass641 642    def origin_information(self):643        if self.relativize:644            effective = dns.name.empty645        else:646            effective = self.origin647        return (self.origin, self.relativize, effective)648 649    def set_rrsets(self, rrsets):650        self.rrsets = rrsets651 652 653def read_rrsets(654    text: Any,655    name: Optional[Union[dns.name.Name, str]] = None,656    ttl: Optional[int] = None,657    rdclass: Optional[Union[dns.rdataclass.RdataClass, str]] = dns.rdataclass.IN,658    default_rdclass: Union[dns.rdataclass.RdataClass, str] = dns.rdataclass.IN,659    rdtype: Optional[Union[dns.rdatatype.RdataType, str]] = None,660    default_ttl: Optional[Union[int, str]] = None,661    idna_codec: Optional[dns.name.IDNACodec] = None,662    origin: Optional[Union[dns.name.Name, str]] = dns.name.root,663    relativize: bool = False,664) -> List[dns.rrset.RRset]:665    """Read one or more rrsets from the specified text, possibly subject666    to restrictions.667 668    *text*, a file object or a string, is the input to process.669 670    *name*, a string, ``dns.name.Name``, or ``None``, is the owner name of671    the rrset.  If not ``None``, then the owner name is "forced", and the672    input must not specify an owner name.  If ``None``, then any owner names673    are allowed and must be present in the input.674 675    *ttl*, an ``int``, string, or None.  If not ``None``, the the TTL is676    forced to be the specified value and the input must not specify a TTL.677    If ``None``, then a TTL may be specified in the input.  If it is not678    specified, then the *default_ttl* will be used.679 680    *rdclass*, a ``dns.rdataclass.RdataClass``, string, or ``None``.  If681    not ``None``, then the class is forced to the specified value, and the682    input must not specify a class.  If ``None``, then the input may specify683    a class that matches *default_rdclass*.  Note that it is not possible to684    return rrsets with differing classes; specifying ``None`` for the class685    simply allows the user to optionally type a class as that may be convenient686    when cutting and pasting.687 688    *default_rdclass*, a ``dns.rdataclass.RdataClass`` or string.  The class689    of the returned rrsets.690 691    *rdtype*, a ``dns.rdatatype.RdataType``, string, or ``None``.  If not692    ``None``, then the type is forced to the specified value, and the693    input must not specify a type.  If ``None``, then a type must be present694    for each RR.695 696    *default_ttl*, an ``int``, string, or ``None``.  If not ``None``, then if697    the TTL is not forced and is not specified, then this value will be used.698    if ``None``, then if the TTL is not forced an error will occur if the TTL699    is not specified.700 701    *idna_codec*, a ``dns.name.IDNACodec``, specifies the IDNA702    encoder/decoder.  If ``None``, the default IDNA 2003 encoder/decoder703    is used.  Note that codecs only apply to the owner name; dnspython does704    not do IDNA for names in rdata, as there is no IDNA zonefile format.705 706    *origin*, a string, ``dns.name.Name``, or ``None``, is the origin for any707    relative names in the input, and also the origin to relativize to if708    *relativize* is ``True``.709 710    *relativize*, a bool.  If ``True``, names are relativized to the *origin*;711    if ``False`` then any relative names in the input are made absolute by712    appending the *origin*.713    """714    if isinstance(origin, str):715        origin = dns.name.from_text(origin, dns.name.root, idna_codec)716    if isinstance(name, str):717        name = dns.name.from_text(name, origin, idna_codec)718    if isinstance(ttl, str):719        ttl = dns.ttl.from_text(ttl)720    if isinstance(default_ttl, str):721        default_ttl = dns.ttl.from_text(default_ttl)722    if rdclass is not None:723        rdclass = dns.rdataclass.RdataClass.make(rdclass)724    else:725        rdclass = None726    default_rdclass = dns.rdataclass.RdataClass.make(default_rdclass)727    if rdtype is not None:728        rdtype = dns.rdatatype.RdataType.make(rdtype)729    else:730        rdtype = None731    manager = RRSetsReaderManager(origin, relativize, default_rdclass)732    with manager.writer(True) as txn:733        tok = dns.tokenizer.Tokenizer(text, "<input>", idna_codec=idna_codec)734        reader = Reader(735            tok,736            default_rdclass,737            txn,738            allow_directives=False,739            force_name=name,740            force_ttl=ttl,741            force_rdclass=rdclass,742            force_rdtype=rdtype,743            default_ttl=default_ttl,744        )745        reader.read()746    return manager.rrsets747