# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2009-2017 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
"""EDNS Options"""
import binascii
import math
import socket
import struct
from typing import Any
import dns.enum
import dns.inet
import dns.ipv4
import dns.ipv6
import dns.name
import dns.rdata
import dns.wire
[docs]
class OptionType(dns.enum.IntEnum):
"""EDNS option type codes"""
#: NSID
NSID = 3
#: DAU
DAU = 5
#: DHU
DHU = 6
#: N3U
N3U = 7
#: ECS (client-subnet)
ECS = 8
#: EXPIRE
EXPIRE = 9
#: COOKIE
COOKIE = 10
#: KEEPALIVE
KEEPALIVE = 11
#: PADDING
PADDING = 12
#: CHAIN
CHAIN = 13
#: EDE (extended-dns-error)
EDE = 15
#: REPORTCHANNEL
REPORTCHANNEL = 18
#: EDE-EXTRA-TEXT-LANGUAGE (Extended DNS Error EXTRA-TEXT language)
EDE_EXTRA_TEXT_LANGUAGE = 22
#: FILTERING-CONTACT
FILTERING_CONTACT = 23
#: FILTERING-ORGANIZATION
FILTERING_ORGANIZATION = 24
#: FILTERING-DB
FILTERING_DB = 25
@classmethod
def _maximum(cls):
return 65535
[docs]
class Option:
"""Base class for all EDNS option types."""
def __init__(self, otype: OptionType | str):
"""Initialize an option.
:param otype: The option type.
:type otype: :py:class:`dns.edns.OptionType`
"""
self.otype = OptionType.make(otype)
[docs]
def to_wire(self, file: Any | None = None) -> bytes | None:
"""Convert an option to wire format.
:rtype: bytes or ``None``
"""
raise NotImplementedError # pragma: no cover
def to_text(self) -> str:
raise NotImplementedError # pragma: no cover
[docs]
def to_generic(self) -> "GenericOption":
"""Create a :py:class:`dns.edns.GenericOption` equivalent of this option.
:rtype: :py:class:`dns.edns.GenericOption`
"""
wire = self.to_wire()
assert wire is not None # for mypy
return GenericOption(self.otype, wire)
[docs]
@classmethod
def from_wire_parser(cls, otype: OptionType, parser: "dns.wire.Parser") -> "Option":
"""Build an EDNS option object from wire format.
:param otype: The option type.
:type otype: :py:class:`dns.edns.OptionType`
:param parser: The parser, restricted to the option length.
:type parser: :py:class:`dns.wire.Parser`
:rtype: :py:class:`dns.edns.Option`
"""
raise NotImplementedError # pragma: no cover
def _cmp(self, other):
"""Compare an EDNS option with another option of the same type.
Returns < 0 if < *other*, 0 if == *other*, and > 0 if > *other*.
"""
wire = self.to_wire()
owire = other.to_wire()
if wire == owire:
return 0
if wire > owire:
return 1
return -1
def __eq__(self, other):
if not isinstance(other, Option):
return False
if self.otype != other.otype:
return False
return self._cmp(other) == 0
def __ne__(self, other):
if not isinstance(other, Option):
return True
if self.otype != other.otype:
return True
return self._cmp(other) != 0
def __lt__(self, other):
if not isinstance(other, Option) or self.otype != other.otype:
return NotImplemented
return self._cmp(other) < 0
def __le__(self, other):
if not isinstance(other, Option) or self.otype != other.otype:
return NotImplemented
return self._cmp(other) <= 0
def __ge__(self, other):
if not isinstance(other, Option) or self.otype != other.otype:
return NotImplemented
return self._cmp(other) >= 0
def __gt__(self, other):
if not isinstance(other, Option) or self.otype != other.otype:
return NotImplemented
return self._cmp(other) > 0
def __str__(self):
return self.to_text()
[docs]
class GenericOption(Option): # lgtm[py/missing-equals]
"""Generic Option Class
This class is used for EDNS option types for which we have no better
implementation.
"""
def __init__(self, otype: OptionType | str, data: bytes | str):
super().__init__(otype)
self.data = dns.rdata.Rdata._as_bytes(data, True)
[docs]
def to_wire(self, file: Any | None = None) -> bytes | None:
if file:
file.write(self.data)
return None
else:
return self.data
def to_text(self) -> str:
return f"Generic {self.otype}"
[docs]
def to_generic(self) -> "GenericOption":
return self
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: "dns.wire.Parser"
) -> Option:
return cls(otype, parser.get_remaining())
[docs]
class ECSOption(Option): # lgtm[py/missing-equals]
"""EDNS Client Subnet (ECS, RFC7871)"""
def __init__(self, address: str, srclen: int | None = None, scopelen: int = 0):
"""Initialize an ECSOption.
:param address: The client address information.
:type address: str
:param srclen: The source prefix length (leftmost number of bits of the
address to be used for the lookup). Defaults to 24 for IPv4 and 56
for IPv6.
:type srclen: int or ``None``
:param scopelen: The scope prefix length. Must be 0 in queries; should
be set in responses.
:type scopelen: int
"""
super().__init__(OptionType.ECS)
af = dns.inet.af_for_address(address)
if af == socket.AF_INET6:
self.family = 2
if srclen is None:
srclen = 56
address = dns.rdata.Rdata._as_ipv6_address(address)
srclen = dns.rdata.Rdata._as_int(srclen, 0, 128)
scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 128)
elif af == socket.AF_INET:
self.family = 1
if srclen is None:
srclen = 24
address = dns.rdata.Rdata._as_ipv4_address(address)
srclen = dns.rdata.Rdata._as_int(srclen, 0, 32)
scopelen = dns.rdata.Rdata._as_int(scopelen, 0, 32)
else: # pragma: no cover (this will never happen)
raise ValueError("Bad address family")
assert srclen is not None
self.address = address
self.srclen = srclen
self.scopelen = scopelen
addrdata = dns.inet.inet_pton(af, address)
nbytes = int(math.ceil(srclen / 8.0))
# Truncate to srclen and pad to the end of the last octet needed
# See RFC section 6
self.addrdata = addrdata[:nbytes]
nbits = srclen % 8
if nbits != 0:
last = struct.pack("B", ord(self.addrdata[-1:]) & (0xFF << (8 - nbits)))
self.addrdata = self.addrdata[:-1] + last
def to_text(self) -> str:
return f"ECS {self.address}/{self.srclen} scope/{self.scopelen}"
[docs]
@staticmethod
def from_text(text: str) -> Option:
"""Convert a string into a `dns.edns.ECSOption`
*text*, a `str`, the text form of the option.
Returns a `dns.edns.ECSOption`.
Examples:
>>> import dns.edns
>>>
>>> # basic example
>>> dns.edns.ECSOption.from_text('1.2.3.4/24')
>>>
>>> # also understands scope
>>> dns.edns.ECSOption.from_text('1.2.3.4/24/32')
>>>
>>> # IPv6
>>> dns.edns.ECSOption.from_text('2001:4b98::1/64/64')
>>>
>>> # it understands results from `dns.edns.ECSOption.to_text()`
>>> dns.edns.ECSOption.from_text('ECS 1.2.3.4/24/32')
"""
optional_prefix = "ECS"
tokens = text.split()
ecs_text = None
if len(tokens) == 1:
ecs_text = tokens[0]
elif len(tokens) == 2:
if tokens[0] != optional_prefix:
raise ValueError(f'could not parse ECS from "{text}"')
ecs_text = tokens[1]
else:
raise ValueError(f'could not parse ECS from "{text}"')
n_slashes = ecs_text.count("/")
if n_slashes == 1:
address, tsrclen = ecs_text.split("/")
tscope = "0"
elif n_slashes == 2:
address, tsrclen, tscope = ecs_text.split("/")
else:
raise ValueError(f'could not parse ECS from "{text}"')
try:
scope = int(tscope)
except ValueError:
raise ValueError("invalid scope " + f'"{tscope}": scope must be an integer')
try:
srclen = int(tsrclen)
except ValueError:
raise ValueError(
"invalid srclen " + f'"{tsrclen}": srclen must be an integer'
)
return ECSOption(address, srclen, scope)
[docs]
def to_wire(self, file: Any | None = None) -> bytes | None:
value = (
struct.pack("!HBB", self.family, self.srclen, self.scopelen) + self.addrdata
)
if file:
file.write(value)
return None
else:
return value
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: "dns.wire.Parser"
) -> Option:
family, src, scope = parser.get_struct("!HBB")
addrlen = int(math.ceil(src / 8.0))
prefix = parser.get_bytes(addrlen)
if family == 1:
pad = 4 - addrlen
addr = dns.ipv4.inet_ntoa(prefix + b"\x00" * pad)
elif family == 2:
pad = 16 - addrlen
addr = dns.ipv6.inet_ntoa(prefix + b"\x00" * pad)
else:
raise ValueError("unsupported family")
return cls(addr, src, scope)
[docs]
class EDECode(dns.enum.IntEnum):
"""Extended DNS Error (EDE) codes"""
OTHER = 0
UNSUPPORTED_DNSKEY_ALGORITHM = 1
UNSUPPORTED_DS_DIGEST_TYPE = 2
STALE_ANSWER = 3
FORGED_ANSWER = 4
DNSSEC_INDETERMINATE = 5
DNSSEC_BOGUS = 6
SIGNATURE_EXPIRED = 7
SIGNATURE_NOT_YET_VALID = 8
DNSKEY_MISSING = 9
RRSIGS_MISSING = 10
NO_ZONE_KEY_BIT_SET = 11
NSEC_MISSING = 12
CACHED_ERROR = 13
NOT_READY = 14
BLOCKED = 15
CENSORED = 16
FILTERED = 17
PROHIBITED = 18
STALE_NXDOMAIN_ANSWER = 19
NOT_AUTHORITATIVE = 20
NOT_SUPPORTED = 21
NO_REACHABLE_AUTHORITY = 22
NETWORK_ERROR = 23
INVALID_DATA = 24
SIGNATURE_EXPIRED_BEFORE_VALID = 25
TOO_EARLY = 26
UNSUPPORTED_NSEC3_ITERATIONS_VALUE = 27
UNABLE_TO_CONFORM_TO_POLICY = 28
SYNTHESIZED = 29
INVALID_QUERY_TYPE = 30
RATE_LIMITED = 31
OVER_QUOTA = 32
@classmethod
def _maximum(cls):
return 65535
[docs]
class EDEOption(Option): # lgtm[py/missing-equals]
"""Extended DNS Error (EDE, RFC8914)"""
_preserve_case = {"DNSKEY", "DS", "DNSSEC", "RRSIGs", "NSEC", "NXDOMAIN"}
def __init__(self, code: EDECode | str, text: str | None = None):
"""Initialize an EDEOption.
:param code: The info code of the extended error.
:type code: :py:class:`dns.edns.EDECode` or str
:param text: Additional information about the error.
:type text: str or ``None``
"""
super().__init__(OptionType.EDE)
self.code = EDECode.make(code)
if text is not None and not isinstance(text, str):
raise ValueError("text must be string or None")
self.text = text
def to_text(self) -> str:
output = f"EDE {self.code}"
if self.code in EDECode:
desc = EDECode.to_text(self.code)
desc = " ".join(
word if word in self._preserve_case else word.title()
for word in desc.split("_")
)
output += f" ({desc})"
if self.text is not None:
output += f": {self.text}"
return output
[docs]
def to_wire(self, file: Any | None = None) -> bytes | None:
value = struct.pack("!H", self.code)
if self.text is not None:
value += self.text.encode("utf8")
if file:
file.write(value)
return None
else:
return value
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: "dns.wire.Parser"
) -> Option:
code = EDECode.make(parser.get_uint16())
text = parser.get_remaining()
if text:
if text[-1] == 0: # text MAY be null-terminated
text = text[:-1]
btext = text.decode("utf8")
else:
btext = None
return cls(code, btext)
[docs]
class NSIDOption(Option):
def __init__(self, nsid: bytes):
super().__init__(OptionType.NSID)
self.nsid = nsid
[docs]
def to_wire(self, file: Any = None) -> bytes | None:
if file:
file.write(self.nsid)
return None
else:
return self.nsid
def to_text(self) -> str:
if all(c >= 0x20 and c <= 0x7E for c in self.nsid):
# All ASCII printable, so it's probably a string.
value = self.nsid.decode()
else:
value = binascii.hexlify(self.nsid).decode()
return f"NSID {value}"
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: dns.wire.Parser
) -> Option:
return cls(parser.get_remaining())
[docs]
class CookieOption(Option):
def __init__(self, client: bytes, server: bytes):
super().__init__(OptionType.COOKIE)
self.client = client
self.server = server
if len(client) != 8:
raise ValueError("client cookie must be 8 bytes")
if len(server) != 0 and (len(server) < 8 or len(server) > 32):
raise ValueError("server cookie must be empty or between 8 and 32 bytes")
[docs]
def to_wire(self, file: Any = None) -> bytes | None:
if file:
file.write(self.client)
if len(self.server) > 0:
file.write(self.server)
return None
else:
return self.client + self.server
def to_text(self) -> str:
client = binascii.hexlify(self.client).decode()
if len(self.server) > 0:
server = binascii.hexlify(self.server).decode()
else:
server = ""
return f"COOKIE {client}{server}"
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: dns.wire.Parser
) -> Option:
return cls(parser.get_bytes(8), parser.get_remaining())
[docs]
class ReportChannelOption(Option):
# RFC 9567
def __init__(self, agent_domain: dns.name.Name):
super().__init__(OptionType.REPORTCHANNEL)
self.agent_domain = agent_domain
[docs]
def to_wire(self, file: Any = None) -> bytes | None:
return self.agent_domain.to_wire(file)
def to_text(self) -> str:
return "REPORTCHANNEL " + self.agent_domain.to_text()
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: dns.wire.Parser
) -> Option:
return cls(parser.get_name())
[docs]
class FilteringOrganizationOption(Option):
"""Filtering organization (FILTERING-ORGANIZATION)"""
def __init__(self, organization: str):
"""Initialize a FilteringOrganizationOption.
:param organization: The filtering organization.
:type organization: str
"""
super().__init__(OptionType.FILTERING_ORGANIZATION)
self.organization = organization
[docs]
def to_wire(self, file: Any = None) -> bytes | None:
if file:
file.write(self.organization.encode("utf8"))
return None
else:
return self.organization.encode()
def to_text(self) -> str:
return f"FILTERING-ORGANIZATION {self.organization}"
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: dns.wire.Parser
) -> Option:
return cls(parser.get_remaining().decode("utf8"))
[docs]
class FilteringDBOption(Option):
"""Filtering DB (FILTERING-DB)"""
def __init__(self, db: str):
"""Initialize a FilteringDBOption.
:param db: The filtering database containing the identifier,
name, or description of the filtering database
against which a matched query caused the filtering to
occur.
:type db: str
"""
super().__init__(OptionType.FILTERING_DB)
self.db = db
[docs]
def to_wire(self, file: Any = None) -> bytes | None:
if file:
file.write(self.db.encode("utf8"))
return None
else:
return self.db.encode()
def to_text(self) -> str:
return f"FILTERING-DB {self.db}"
[docs]
@classmethod
def from_wire_parser(
cls, otype: OptionType | str, parser: dns.wire.Parser
) -> Option:
return cls(parser.get_remaining().decode("utf8"))
_type_to_class: dict[OptionType, Any] = {
OptionType.ECS: ECSOption,
OptionType.EDE: EDEOption,
OptionType.NSID: NSIDOption,
OptionType.COOKIE: CookieOption,
OptionType.REPORTCHANNEL: ReportChannelOption,
OptionType.EDE_EXTRA_TEXT_LANGUAGE: EDEExtraTextLanguageOption,
OptionType.FILTERING_CONTACT: FilteringContactOption,
OptionType.FILTERING_ORGANIZATION: FilteringOrganizationOption,
OptionType.FILTERING_DB: FilteringDBOption,
}
[docs]
def get_option_class(otype: OptionType) -> Any:
"""Return the class for the specified option type.
The GenericOption class is used if a more specific class is not
known.
"""
cls = _type_to_class.get(otype)
if cls is None:
cls = GenericOption
return cls
[docs]
def option_from_wire_parser(
otype: OptionType | str, parser: "dns.wire.Parser"
) -> Option:
"""Build an EDNS option object from wire format.
:param otype: The option type.
:type otype: int or :py:class:`dns.edns.OptionType`
:param parser: The parser, restricted to the option length.
:type parser: :py:class:`dns.wire.Parser`
:rtype: :py:class:`dns.edns.Option`
"""
otype = OptionType.make(otype)
cls = get_option_class(otype)
return cls.from_wire_parser(otype, parser)
[docs]
def option_from_wire(
otype: OptionType | str, wire: bytes, current: int, olen: int
) -> Option:
"""Build an EDNS option object from wire format.
:param otype: The option type.
:type otype: int or :py:class:`dns.edns.OptionType`
:param wire: The wire-format message.
:type wire: bytes
:param current: The offset in *wire* of the beginning of the rdata.
:type current: int
:param olen: The length of the wire-format option data.
:type olen: int
:rtype: :py:class:`dns.edns.Option`
"""
parser = dns.wire.Parser(wire, current)
with parser.restrict_to(olen):
return option_from_wire_parser(otype, parser)
[docs]
def register_type(implementation: Any, otype: OptionType) -> None:
"""Register the implementation of an option type.
:param implementation: A subclass of :py:class:`dns.edns.Option`.
:param otype: The option type.
:type otype: :py:class:`dns.edns.OptionType`
"""
_type_to_class[otype] = implementation
### BEGIN generated OptionType constants
NSID = OptionType.NSID
DAU = OptionType.DAU
DHU = OptionType.DHU
N3U = OptionType.N3U
ECS = OptionType.ECS
EXPIRE = OptionType.EXPIRE
COOKIE = OptionType.COOKIE
KEEPALIVE = OptionType.KEEPALIVE
PADDING = OptionType.PADDING
CHAIN = OptionType.CHAIN
EDE = OptionType.EDE
REPORTCHANNEL = OptionType.REPORTCHANNEL
EDE_EXTRA_TEXT_LANGUAGE = OptionType.EDE_EXTRA_TEXT_LANGUAGE
FILTERING_CONTACT = OptionType.FILTERING_CONTACT
FILTERING_ORGANIZATION = OptionType.FILTERING_ORGANIZATION
FILTERING_DB = OptionType.FILTERING_DB
### END generated OptionType constants