codekingpro/portable-devtools

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license2 3# Copyright (C) 2009-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 18"""EDNS Options"""19 20import binascii21import math22import socket23import struct24from typing import Any, Dict, Optional, Union25 26import dns.enum27import dns.inet28import dns.rdata29import dns.wire30 31 32class OptionType(dns.enum.IntEnum):33    #: NSID34    NSID = 335    #: DAU36    DAU = 537    #: DHU38    DHU = 639    #: N3U40    N3U = 741    #: ECS (client-subnet)42    ECS = 843    #: EXPIRE44    EXPIRE = 945    #: COOKIE46    COOKIE = 1047    #: KEEPALIVE48    KEEPALIVE = 1149    #: PADDING50    PADDING = 1251    #: CHAIN52    CHAIN = 1353    #: EDE (extended-dns-error)54    EDE = 1555 56    @classmethod57    def _maximum(cls):58        return 6553559 60 61class Option:62    """Base class for all EDNS option types."""63 64    def __init__(self, otype: Union[OptionType, str]):65        """Initialize an option.66 67        *otype*, a ``dns.edns.OptionType``, is the option type.68        """69        self.otype = OptionType.make(otype)70 71    def to_wire(self, file: Optional[Any] = None) -> Optional[bytes]:72        """Convert an option to wire format.73 74        Returns a ``bytes`` or ``None``.75 76        """77        raise NotImplementedError  # pragma: no cover78 79    def to_text(self) -> str:80        raise NotImplementedError  # pragma: no cover81 82    @classmethod83    def from_wire_parser(cls, otype: OptionType, parser: "dns.wire.Parser") -> "Option":84        """Build an EDNS option object from wire format.85 86        *otype*, a ``dns.edns.OptionType``, is the option type.87 88        *parser*, a ``dns.wire.Parser``, the parser, which should be89        restructed to the option length.90 91        Returns a ``dns.edns.Option``.92        """93        raise NotImplementedError  # pragma: no cover94 95    def _cmp(self, other):96        """Compare an EDNS option with another option of the same type.97 98        Returns < 0 if < *other*, 0 if == *other*, and > 0 if > *other*.99        """100        wire = self.to_wire()101        owire = other.to_wire()102        if wire == owire:103            return 0104        if wire > owire:105            return 1106        return -1107 108    def __eq__(self, other):109        if not isinstance(other, Option):110            return False111        if self.otype != other.otype:112            return False113        return self._cmp(other) == 0114 115    def __ne__(self, other):116        if not isinstance(other, Option):117            return True118        if self.otype != other.otype:119            return True120        return self._cmp(other) != 0121 122    def __lt__(self, other):123        if not isinstance(other, Option) or self.otype != other.otype:124            return NotImplemented125        return self._cmp(other) < 0126 127    def __le__(self, other):128        if not isinstance(other, Option) or self.otype != other.otype:129            return NotImplemented130        return self._cmp(other) <= 0131 132    def __ge__(self, other):133        if not isinstance(other, Option) or self.otype != other.otype:134            return NotImplemented135        return self._cmp(other) >= 0136 137    def __gt__(self, other):138        if not isinstance(other, Option) or self.otype != other.otype:139            return NotImplemented140        return self._cmp(other) > 0141 142    def __str__(self):143        return self.to_text()144 145 146class GenericOption(Option):  # lgtm[py/missing-equals]147    """Generic Option Class148 149    This class is used for EDNS option types for which we have no better150    implementation.151    """152 153    def __init__(self, otype: Union[OptionType, str], data: Union[bytes, str]):154        super().__init__(otype)155        self.data = dns.rdata.Rdata._as_bytes(data, True)156 157    def to_wire(self, file: Optional[Any] = None) -> Optional[bytes]:158        if file:159            file.write(self.data)160            return None161        else:162            return self.data163 164    def to_text(self) -> str:165        return "Generic %d" % self.otype166 167    @classmethod168    def from_wire_parser(169        cls, otype: Union[OptionType, str], parser: "dns.wire.Parser"170    ) -> Option:171        return cls(otype, parser.get_remaining())172 173 174class ECSOption(Option):  # lgtm[py/missing-equals]175    """EDNS Client Subnet (ECS, RFC7871)"""176 177    def __init__(self, address: str, srclen: Optional[int] = None, scopelen: int = 0):178        """*address*, a ``str``, is the client address information.179 180        *srclen*, an ``int``, the source prefix length, which is the181        leftmost number of bits of the address to be used for the182        lookup.  The default is 24 for IPv4 and 56 for IPv6.183 184        *scopelen*, an ``int``, the scope prefix length.  This value185        must be 0 in queries, and should be set in responses.186        """187 188        super().__init__(OptionType.ECS)189        af = dns.inet.af_for_address(address)190 191        if af == socket.AF_INET6:192            self.family = 2193            if srclen is None:194                srclen = 56195            address = dns.rdata.Rdata._as_ipv6_address(address)196            srclen = dns.rdata.Rdata._as_int(srclen, 0, 128)197            scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 128)198        elif af == socket.AF_INET:199            self.family = 1200            if srclen is None:201                srclen = 24202            address = dns.rdata.Rdata._as_ipv4_address(address)203            srclen = dns.rdata.Rdata._as_int(srclen, 0, 32)204            scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 32)205        else:  # pragma: no cover   (this will never happen)206            raise ValueError("Bad address family")207 208        assert srclen is not None209        self.address = address210        self.srclen = srclen211        self.scopelen = scopelen212 213        addrdata = dns.inet.inet_pton(af, address)214        nbytes = int(math.ceil(srclen / 8.0))215 216        # Truncate to srclen and pad to the end of the last octet needed217        # See RFC section 6218        self.addrdata = addrdata[:nbytes]219        nbits = srclen % 8220        if nbits != 0:221            last = struct.pack("B", ord(self.addrdata[-1:]) & (0xFF << (8 - nbits)))222            self.addrdata = self.addrdata[:-1] + last223 224    def to_text(self) -> str:225        return "ECS {}/{} scope/{}".format(self.address, self.srclen, self.scopelen)226 227    @staticmethod228    def from_text(text: str) -> Option:229        """Convert a string into a `dns.edns.ECSOption`230 231        *text*, a `str`, the text form of the option.232 233        Returns a `dns.edns.ECSOption`.234 235        Examples:236 237        >>> import dns.edns238        >>>239        >>> # basic example240        >>> dns.edns.ECSOption.from_text('1.2.3.4/24')241        >>>242        >>> # also understands scope243        >>> dns.edns.ECSOption.from_text('1.2.3.4/24/32')244        >>>245        >>> # IPv6246        >>> dns.edns.ECSOption.from_text('2001:4b98::1/64/64')247        >>>248        >>> # it understands results from `dns.edns.ECSOption.to_text()`249        >>> dns.edns.ECSOption.from_text('ECS 1.2.3.4/24/32')250        """251        optional_prefix = "ECS"252        tokens = text.split()253        ecs_text = None254        if len(tokens) == 1:255            ecs_text = tokens[0]256        elif len(tokens) == 2:257            if tokens[0] != optional_prefix:258                raise ValueError('could not parse ECS from "{}"'.format(text))259            ecs_text = tokens[1]260        else:261            raise ValueError('could not parse ECS from "{}"'.format(text))262        n_slashes = ecs_text.count("/")263        if n_slashes == 1:264            address, tsrclen = ecs_text.split("/")265            tscope = "0"266        elif n_slashes == 2:267            address, tsrclen, tscope = ecs_text.split("/")268        else:269            raise ValueError('could not parse ECS from "{}"'.format(text))270        try:271            scope = int(tscope)272        except ValueError:273            raise ValueError(274                "invalid scope " + '"{}": scope must be an integer'.format(tscope)275            )276        try:277            srclen = int(tsrclen)278        except ValueError:279            raise ValueError(280                "invalid srclen " + '"{}": srclen must be an integer'.format(tsrclen)281            )282        return ECSOption(address, srclen, scope)283 284    def to_wire(self, file: Optional[Any] = None) -> Optional[bytes]:285        value = (286            struct.pack("!HBB", self.family, self.srclen, self.scopelen) + self.addrdata287        )288        if file:289            file.write(value)290            return None291        else:292            return value293 294    @classmethod295    def from_wire_parser(296        cls, otype: Union[OptionType, str], parser: "dns.wire.Parser"297    ) -> Option:298        family, src, scope = parser.get_struct("!HBB")299        addrlen = int(math.ceil(src / 8.0))300        prefix = parser.get_bytes(addrlen)301        if family == 1:302            pad = 4 - addrlen303            addr = dns.ipv4.inet_ntoa(prefix + b"\x00" * pad)304        elif family == 2:305            pad = 16 - addrlen306            addr = dns.ipv6.inet_ntoa(prefix + b"\x00" * pad)307        else:308            raise ValueError("unsupported family")309 310        return cls(addr, src, scope)311 312 313class EDECode(dns.enum.IntEnum):314    OTHER = 0315    UNSUPPORTED_DNSKEY_ALGORITHM = 1316    UNSUPPORTED_DS_DIGEST_TYPE = 2317    STALE_ANSWER = 3318    FORGED_ANSWER = 4319    DNSSEC_INDETERMINATE = 5320    DNSSEC_BOGUS = 6321    SIGNATURE_EXPIRED = 7322    SIGNATURE_NOT_YET_VALID = 8323    DNSKEY_MISSING = 9324    RRSIGS_MISSING = 10325    NO_ZONE_KEY_BIT_SET = 11326    NSEC_MISSING = 12327    CACHED_ERROR = 13328    NOT_READY = 14329    BLOCKED = 15330    CENSORED = 16331    FILTERED = 17332    PROHIBITED = 18333    STALE_NXDOMAIN_ANSWER = 19334    NOT_AUTHORITATIVE = 20335    NOT_SUPPORTED = 21336    NO_REACHABLE_AUTHORITY = 22337    NETWORK_ERROR = 23338    INVALID_DATA = 24339 340    @classmethod341    def _maximum(cls):342        return 65535343 344 345class EDEOption(Option):  # lgtm[py/missing-equals]346    """Extended DNS Error (EDE, RFC8914)"""347 348    _preserve_case = {"DNSKEY", "DS", "DNSSEC", "RRSIGs", "NSEC", "NXDOMAIN"}349 350    def __init__(self, code: Union[EDECode, str], text: Optional[str] = None):351        """*code*, a ``dns.edns.EDECode`` or ``str``, the info code of the352        extended error.353 354        *text*, a ``str`` or ``None``, specifying additional information about355        the error.356        """357 358        super().__init__(OptionType.EDE)359 360        self.code = EDECode.make(code)361        if text is not None and not isinstance(text, str):362            raise ValueError("text must be string or None")363        self.text = text364 365    def to_text(self) -> str:366        output = f"EDE {self.code}"367        if self.code in EDECode:368            desc = EDECode.to_text(self.code)369            desc = " ".join(370                word if word in self._preserve_case else word.title()371                for word in desc.split("_")372            )373            output += f" ({desc})"374        if self.text is not None:375            output += f": {self.text}"376        return output377 378    def to_wire(self, file: Optional[Any] = None) -> Optional[bytes]:379        value = struct.pack("!H", self.code)380        if self.text is not None:381            value += self.text.encode("utf8")382 383        if file:384            file.write(value)385            return None386        else:387            return value388 389    @classmethod390    def from_wire_parser(391        cls, otype: Union[OptionType, str], parser: "dns.wire.Parser"392    ) -> Option:393        code = EDECode.make(parser.get_uint16())394        text = parser.get_remaining()395 396        if text:397            if text[-1] == 0:  # text MAY be null-terminated398                text = text[:-1]399            btext = text.decode("utf8")400        else:401            btext = None402 403        return cls(code, btext)404 405 406class NSIDOption(Option):407    def __init__(self, nsid: bytes):408        super().__init__(OptionType.NSID)409        self.nsid = nsid410 411    def to_wire(self, file: Any = None) -> Optional[bytes]:412        if file:413            file.write(self.nsid)414            return None415        else:416            return self.nsid417 418    def to_text(self) -> str:419        if all(c >= 0x20 and c <= 0x7E for c in self.nsid):420            # All ASCII printable, so it's probably a string.421            value = self.nsid.decode()422        else:423            value = binascii.hexlify(self.nsid).decode()424        return f"NSID {value}"425 426    @classmethod427    def from_wire_parser(428        cls, otype: Union[OptionType, str], parser: dns.wire.Parser429    ) -> Option:430        return cls(parser.get_remaining())431 432 433_type_to_class: Dict[OptionType, Any] = {434    OptionType.ECS: ECSOption,435    OptionType.EDE: EDEOption,436    OptionType.NSID: NSIDOption,437}438 439 440def get_option_class(otype: OptionType) -> Any:441    """Return the class for the specified option type.442 443    The GenericOption class is used if a more specific class is not444    known.445    """446 447    cls = _type_to_class.get(otype)448    if cls is None:449        cls = GenericOption450    return cls451 452 453def option_from_wire_parser(454    otype: Union[OptionType, str], parser: "dns.wire.Parser"455) -> Option:456    """Build an EDNS option object from wire format.457 458    *otype*, an ``int``, is the option type.459 460    *parser*, a ``dns.wire.Parser``, the parser, which should be461    restricted to the option length.462 463    Returns an instance of a subclass of ``dns.edns.Option``.464    """465    otype = OptionType.make(otype)466    cls = get_option_class(otype)467    return cls.from_wire_parser(otype, parser)468 469 470def option_from_wire(471    otype: Union[OptionType, str], wire: bytes, current: int, olen: int472) -> Option:473    """Build an EDNS option object from wire format.474 475    *otype*, an ``int``, is the option type.476 477    *wire*, a ``bytes``, is the wire-format message.478 479    *current*, an ``int``, is the offset in *wire* of the beginning480    of the rdata.481 482    *olen*, an ``int``, is the length of the wire-format option data483 484    Returns an instance of a subclass of ``dns.edns.Option``.485    """486    parser = dns.wire.Parser(wire, current)487    with parser.restrict_to(olen):488        return option_from_wire_parser(otype, parser)489 490 491def register_type(implementation: Any, otype: OptionType) -> None:492    """Register the implementation of an option type.493 494    *implementation*, a ``class``, is a subclass of ``dns.edns.Option``.495 496    *otype*, an ``int``, is the option type.497    """498 499    _type_to_class[otype] = implementation500 501 502### BEGIN generated OptionType constants503 504NSID = OptionType.NSID505DAU = OptionType.DAU506DHU = OptionType.DHU507N3U = OptionType.N3U508ECS = OptionType.ECS509EXPIRE = OptionType.EXPIRE510COOKIE = OptionType.COOKIE511KEEPALIVE = OptionType.KEEPALIVE512PADDING = OptionType.PADDING513CHAIN = OptionType.CHAIN514EDE = OptionType.EDE515 516### END generated OptionType constants517