codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3# Copyright (C) 2003-2017 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 18from typing import Any, List, Optional, Tuple, Union19 20import dns.exception21import dns.message22import dns.name23import dns.rcode24import dns.rdataset25import dns.rdatatype26import dns.serial27import dns.transaction28import dns.tsig29import dns.zone30 31 32class TransferError(dns.exception.DNSException):33    """A zone transfer response got a non-zero rcode."""34 35    def __init__(self, rcode):36        message = "Zone transfer error: %s" % dns.rcode.to_text(rcode)37        super().__init__(message)38        self.rcode = rcode39 40 41class SerialWentBackwards(dns.exception.FormError):42    """The current serial number is less than the serial we know."""43 44 45class UseTCP(dns.exception.DNSException):46    """This IXFR cannot be completed with UDP."""47 48 49class Inbound:50    """51    State machine for zone transfers.52    """53 54    def __init__(55        self,56        txn_manager: dns.transaction.TransactionManager,57        rdtype: dns.rdatatype.RdataType = dns.rdatatype.AXFR,58        serial: Optional[int] = None,59        is_udp: bool = False,60    ):61        """Initialize an inbound zone transfer.62 63        *txn_manager* is a :py:class:`dns.transaction.TransactionManager`.64 65        *rdtype* can be `dns.rdatatype.AXFR` or `dns.rdatatype.IXFR`66 67        *serial* is the base serial number for IXFRs, and is required in68        that case.69 70        *is_udp*, a ``bool`` indidicates if UDP is being used for this71        XFR.72        """73        self.txn_manager = txn_manager74        self.txn: Optional[dns.transaction.Transaction] = None75        self.rdtype = rdtype76        if rdtype == dns.rdatatype.IXFR:77            if serial is None:78                raise ValueError("a starting serial must be supplied for IXFRs")79        elif is_udp:80            raise ValueError("is_udp specified for AXFR")81        self.serial = serial82        self.is_udp = is_udp83        (_, _, self.origin) = txn_manager.origin_information()84        self.soa_rdataset: Optional[dns.rdataset.Rdataset] = None85        self.done = False86        self.expecting_SOA = False87        self.delete_mode = False88 89    def process_message(self, message: dns.message.Message) -> bool:90        """Process one message in the transfer.91 92        The message should have the same relativization as was specified when93        the `dns.xfr.Inbound` was created.  The message should also have been94        created with `one_rr_per_rrset=True` because order matters.95 96        Returns `True` if the transfer is complete, and `False` otherwise.97        """98        if self.txn is None:99            replacement = self.rdtype == dns.rdatatype.AXFR100            self.txn = self.txn_manager.writer(replacement)101        rcode = message.rcode()102        if rcode != dns.rcode.NOERROR:103            raise TransferError(rcode)104        #105        # We don't require a question section, but if it is present is106        # should be correct.107        #108        if len(message.question) > 0:109            if message.question[0].name != self.origin:110                raise dns.exception.FormError("wrong question name")111            if message.question[0].rdtype != self.rdtype:112                raise dns.exception.FormError("wrong question rdatatype")113        answer_index = 0114        if self.soa_rdataset is None:115            #116            # This is the first message.  We're expecting an SOA at117            # the origin.118            #119            if not message.answer or message.answer[0].name != self.origin:120                raise dns.exception.FormError("No answer or RRset not for zone origin")121            rrset = message.answer[0]122            rdataset = rrset123            if rdataset.rdtype != dns.rdatatype.SOA:124                raise dns.exception.FormError("first RRset is not an SOA")125            answer_index = 1126            self.soa_rdataset = rdataset.copy()127            if self.rdtype == dns.rdatatype.IXFR:128                if self.soa_rdataset[0].serial == self.serial:129                    #130                    # We're already up-to-date.131                    #132                    self.done = True133                elif dns.serial.Serial(self.soa_rdataset[0].serial) < self.serial:134                    # It went backwards!135                    raise SerialWentBackwards136                else:137                    if self.is_udp and len(message.answer[answer_index:]) == 0:138                        #139                        # There are no more records, so this is the140                        # "truncated" response.  Say to use TCP141                        #142                        raise UseTCP143                    #144                    # Note we're expecting another SOA so we can detect145                    # if this IXFR response is an AXFR-style response.146                    #147                    self.expecting_SOA = True148        #149        # Process the answer section (other than the initial SOA in150        # the first message).151        #152        for rrset in message.answer[answer_index:]:153            name = rrset.name154            rdataset = rrset155            if self.done:156                raise dns.exception.FormError("answers after final SOA")157            assert self.txn is not None  # for mypy158            if rdataset.rdtype == dns.rdatatype.SOA and name == self.origin:159                #160                # Every time we see an origin SOA delete_mode inverts161                #162                if self.rdtype == dns.rdatatype.IXFR:163                    self.delete_mode = not self.delete_mode164                #165                # If this SOA Rdataset is equal to the first we saw166                # then we're finished. If this is an IXFR we also167                # check that we're seeing the record in the expected168                # part of the response.169                #170                if rdataset == self.soa_rdataset and (171                    self.rdtype == dns.rdatatype.AXFR172                    or (self.rdtype == dns.rdatatype.IXFR and self.delete_mode)173                ):174                    #175                    # This is the final SOA176                    #177                    if self.expecting_SOA:178                        # We got an empty IXFR sequence!179                        raise dns.exception.FormError("empty IXFR sequence")180                    if (181                        self.rdtype == dns.rdatatype.IXFR182                        and self.serial != rdataset[0].serial183                    ):184                        raise dns.exception.FormError("unexpected end of IXFR sequence")185                    self.txn.replace(name, rdataset)186                    self.txn.commit()187                    self.txn = None188                    self.done = True189                else:190                    #191                    # This is not the final SOA192                    #193                    self.expecting_SOA = False194                    if self.rdtype == dns.rdatatype.IXFR:195                        if self.delete_mode:196                            # This is the start of an IXFR deletion set197                            if rdataset[0].serial != self.serial:198                                raise dns.exception.FormError(199                                    "IXFR base serial mismatch"200                                )201                        else:202                            # This is the start of an IXFR addition set203                            self.serial = rdataset[0].serial204                            self.txn.replace(name, rdataset)205                    else:206                        # We saw a non-final SOA for the origin in an AXFR.207                        raise dns.exception.FormError("unexpected origin SOA in AXFR")208                continue209            if self.expecting_SOA:210                #211                # We made an IXFR request and are expecting another212                # SOA RR, but saw something else, so this must be an213                # AXFR response.214                #215                self.rdtype = dns.rdatatype.AXFR216                self.expecting_SOA = False217                self.delete_mode = False218                self.txn.rollback()219                self.txn = self.txn_manager.writer(True)220                #221                # Note we are falling through into the code below222                # so whatever rdataset this was gets written.223                #224            # Add or remove the data225            if self.delete_mode:226                self.txn.delete_exact(name, rdataset)227            else:228                self.txn.add(name, rdataset)229        if self.is_udp and not self.done:230            #231            # This is a UDP IXFR and we didn't get to done, and we didn't232            # get the proper "truncated" response233            #234            raise dns.exception.FormError("unexpected end of UDP IXFR")235        return self.done236 237    #238    # Inbounds are context managers.239    #240 241    def __enter__(self):242        return self243 244    def __exit__(self, exc_type, exc_val, exc_tb):245        if self.txn:246            self.txn.rollback()247        return False248 249 250def make_query(251    txn_manager: dns.transaction.TransactionManager,252    serial: Optional[int] = 0,253    use_edns: Optional[Union[int, bool]] = None,254    ednsflags: Optional[int] = None,255    payload: Optional[int] = None,256    request_payload: Optional[int] = None,257    options: Optional[List[dns.edns.Option]] = None,258    keyring: Any = None,259    keyname: Optional[dns.name.Name] = None,260    keyalgorithm: Union[dns.name.Name, str] = dns.tsig.default_algorithm,261) -> Tuple[dns.message.QueryMessage, Optional[int]]:262    """Make an AXFR or IXFR query.263 264    *txn_manager* is a ``dns.transaction.TransactionManager``, typically a265    ``dns.zone.Zone``.266 267    *serial* is an ``int`` or ``None``.  If 0, then IXFR will be268    attempted using the most recent serial number from the269    *txn_manager*; it is the caller's responsibility to ensure there270    are no write transactions active that could invalidate the271    retrieved serial.  If a serial cannot be determined, AXFR will be272    forced.  Other integer values are the starting serial to use.273    ``None`` forces an AXFR.274 275    Please see the documentation for :py:func:`dns.message.make_query` and276    :py:func:`dns.message.Message.use_tsig` for details on the other parameters277    to this function.278 279    Returns a `(query, serial)` tuple.280    """281    (zone_origin, _, origin) = txn_manager.origin_information()282    if zone_origin is None:283        raise ValueError("no zone origin")284    if serial is None:285        rdtype = dns.rdatatype.AXFR286    elif not isinstance(serial, int):287        raise ValueError("serial is not an integer")288    elif serial == 0:289        with txn_manager.reader() as txn:290            rdataset = txn.get(origin, "SOA")291            if rdataset:292                serial = rdataset[0].serial293                rdtype = dns.rdatatype.IXFR294            else:295                serial = None296                rdtype = dns.rdatatype.AXFR297    elif serial > 0 and serial < 4294967296:298        rdtype = dns.rdatatype.IXFR299    else:300        raise ValueError("serial out-of-range")301    rdclass = txn_manager.get_class()302    q = dns.message.make_query(303        zone_origin,304        rdtype,305        rdclass,306        use_edns,307        False,308        ednsflags,309        payload,310        request_payload,311        options,312    )313    if serial is not None:314        rdata = dns.rdata.from_text(rdclass, "SOA", f". . {serial} 0 0 0 0")315        rrset = q.find_rrset(316            q.authority, zone_origin, rdclass, dns.rdatatype.SOA, create=True317        )318        rrset.add(rdata, 0)319    if keyring is not None:320        q.use_tsig(keyring, keyname, algorithm=keyalgorithm)321    return (q, serial)322 323 324def extract_serial_from_query(query: dns.message.Message) -> Optional[int]:325    """Extract the SOA serial number from query if it is an IXFR and return326    it, otherwise return None.327 328    *query* is a dns.message.QueryMessage that is an IXFR or AXFR request.329 330    Raises if the query is not an IXFR or AXFR, or if an IXFR doesn't have331    an appropriate SOA RRset in the authority section.332    """333    if not isinstance(query, dns.message.QueryMessage):334        raise ValueError("query not a QueryMessage")335    question = query.question[0]336    if question.rdtype == dns.rdatatype.AXFR:337        return None338    elif question.rdtype != dns.rdatatype.IXFR:339        raise ValueError("query is not an AXFR or IXFR")340    soa = query.find_rrset(341        query.authority, question.name, question.rdclass, dns.rdatatype.SOA342    )343    return soa[0].serial344