# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
import base64
import enum
import re
import struct
from typing import Any, TypeVar
import dns._text_util
import dns.enum
import dns.exception
import dns.immutable
import dns.ipv4
import dns.ipv6
import dns.name
import dns.rdata
import dns.rdtypes.util
import dns.renderer
import dns.tokenizer
import dns.wire
# Until there is an RFC, this module is experimental and may be changed in
# incompatible ways.
class UnknownParamKey(dns.exception.DNSException):
"""Unknown SVCB ParamKey"""
[docs]
class ParamKey(dns.enum.IntEnum):
"""SVCB ParamKey"""
MANDATORY = 0
ALPN = 1
NO_DEFAULT_ALPN = 2
PORT = 3
IPV4HINT = 4
ECH = 5
IPV6HINT = 6
DOHPATH = 7
OHTTP = 8
DOCPATH = 10
@classmethod
def _maximum(cls):
return 65535
@classmethod
def _short_name(cls):
return "SVCBParamKey"
@classmethod
def _prefix(cls):
return "KEY"
@classmethod
def _unknown_exception_class(cls):
return UnknownParamKey
class Emptiness(enum.IntEnum):
NEVER = 0
ALWAYS = 1
ALLOWED = 2
def _validate_key(key):
force_generic = False
if isinstance(key, bytes):
# We decode to latin-1 so we get 0-255 as valid and do NOT interpret
# UTF-8 sequences
key = key.decode("latin-1")
if isinstance(key, str):
if key.lower().startswith("key"):
force_generic = True
if key[3:].startswith("0") and len(key) != 4:
# key has leading zeros
raise ValueError("leading zeros in key")
key = key.replace("-", "_")
return (ParamKey.make(key), force_generic)
# Presentation format of a SvcParamKey (RFC 9460 Section 2.1).
_key_text_re = re.compile("[a-z0-9-]{1,63}")
def _check_key_text(key: str) -> None:
if not _key_text_re.fullmatch(key):
raise dns.exception.SyntaxError(f'invalid key "{key}"')
def key_to_text(key):
return ParamKey.to_text(key).replace("_", "-").lower()
# Escape the characters which are special in a comma-separated list; the
# character-string escaping (dns.rdata._escapify) is applied on top of this.
_escaped = b",\\"
def _escapify(qstring: bytes) -> bytes:
text = b""
for c in qstring:
if c in _escaped:
text += b"\\"
text += b"%c" % (c)
return text
def _unescape(value: str) -> bytes:
if value == "":
return b""
unescaped = b""
l = len(value)
i = 0
while i < l:
c = value[i]
i += 1
if c == "\\":
if i >= l: # pragma: no cover (can't happen via tokenizer get())
raise dns.exception.UnexpectedEnd
c = value[i]
i += 1
if dns._text_util.is_ascii_digit(c):
if i >= l:
raise dns.exception.UnexpectedEnd
c2 = value[i]
i += 1
if i >= l:
raise dns.exception.UnexpectedEnd
c3 = value[i]
i += 1
if not (
dns._text_util.is_ascii_digit(c2)
and dns._text_util.is_ascii_digit(c3)
):
raise dns.exception.SyntaxError
codepoint = int(c) * 100 + int(c2) * 10 + int(c3)
if codepoint > 255:
raise dns.exception.SyntaxError
unescaped += b"%c" % (codepoint)
continue
unescaped += c.encode()
return unescaped
def _split(value):
l = len(value)
i = 0
items = []
unescaped = b""
while i < l:
c = value[i]
i += 1
if c == ord("\\"):
if i >= l:
raise dns.exception.UnexpectedEnd
c = value[i]
i += 1
# Only "\," and "\\" are valid (RFC 9460 Appendix A.1).
if c not in _escaped:
raise dns.exception.SyntaxError("invalid escape in value list")
unescaped += b"%c" % (c)
elif c == ord(","):
items.append(unescaped)
unescaped = b""
else:
unescaped += b"%c" % (c)
items.append(unescaped)
return items
[docs]
@dns.immutable.immutable
class Param:
"""Abstract base class for SVCB parameters"""
@classmethod
def emptiness(cls) -> Emptiness:
return Emptiness.NEVER
[docs]
@dns.immutable.immutable
class GenericParam(Param):
"""Generic SVCB parameter"""
def __init__(self, value):
self.value = dns.rdata.Rdata._as_bytes(value, True)
@classmethod
def emptiness(cls):
return Emptiness.ALLOWED
@classmethod
def from_value(cls, value):
if value is None or len(value) == 0:
return None
else:
return cls(_unescape(value))
def to_text(self):
return '"' + dns.rdata._escapify(self.value) + '"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
value = parser.get_bytes(parser.remaining())
if len(value) == 0:
return None
else:
return cls(value)
def to_wire(self, file, origin=None): # pylint: disable=W0613
file.write(self.value)
[docs]
@dns.immutable.immutable
class MandatoryParam(Param):
def __init__(self, keys):
# check for duplicates
keys = sorted([_validate_key(key)[0] for key in keys])
if len(keys) == 0:
raise ValueError("empty mandatory list")
prior_k = None
for k in keys:
if k == prior_k:
raise ValueError(f"duplicate key {k:d}")
prior_k = k
if k == ParamKey.MANDATORY:
raise ValueError("listed the mandatory key as mandatory")
self.keys = tuple(keys)
@classmethod
def from_value(cls, value):
if "\\" in value:
raise ValueError("escape in mandatory value")
keys = value.split(",")
for key in keys:
_check_key_text(key)
return cls([key.encode() for key in keys])
def to_text(self):
return '"' + ",".join([key_to_text(key) for key in self.keys]) + '"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
keys = []
last_key = -1
while parser.remaining() > 0:
key = parser.get_uint16()
if key <= last_key:
raise dns.exception.FormError(
"mandatory keys not in strictly increasing order"
)
last_key = key
keys.append(key)
return cls(keys)
def to_wire(self, file, origin=None): # pylint: disable=W0613
for key in self.keys:
file.write(struct.pack("!H", key))
@dns.immutable.immutable
class _StringList(Param):
"""A comma separated list of strings, possibly empty."""
def __init__(self, ids):
self.ids = dns.rdata.Rdata._as_tuple(
ids, lambda x: dns.rdata.Rdata._as_bytes(x, True, 255, False)
)
@classmethod
def emptiness(cls) -> Emptiness:
return Emptiness.ALLOWED
@classmethod
def from_value(cls, value):
if value is None or value == "":
return None
return cls(_split(_unescape(value)))
def to_text(self):
value = b",".join([_escapify(id) for id in self.ids])
return '"' + dns.rdata._escapify(value) + '"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
if parser.remaining() == 0:
return None
ids = []
while parser.remaining() > 0:
id = parser.get_counted_bytes()
ids.append(id)
return cls(ids)
def to_wire(self, file, origin=None): # pylint: disable=W0613
for id in self.ids:
file.write(struct.pack("!B", len(id)))
file.write(id)
[docs]
@dns.immutable.immutable
class ALPNParam(_StringList):
"""The alpn parameter."""
def __init__(self, ids):
self.ids = dns.rdata.Rdata._as_tuple(
ids, lambda x: dns.rdata.Rdata._as_bytes(x, True, 255, False)
)
if len(self.ids) == 0:
raise ValueError("empty alpn list")
@classmethod
def emptiness(cls):
return Emptiness.NEVER
@dns.immutable.immutable
class NoDefaultALPNParam(Param):
# We don't ever expect to instantiate this class, but we need
# a from_value() and a from_wire_parser(), so we just return None
# from the class methods when things are OK.
@classmethod
def emptiness(cls):
return Emptiness.ALWAYS
@classmethod
def from_value(cls, value):
if value is None or value == "":
return None
else:
raise ValueError("no-default-alpn with non-empty value")
def to_text(self):
raise NotImplementedError # pragma: no cover
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
if parser.remaining() != 0:
raise dns.exception.FormError
return None
def to_wire(self, file, origin=None): # pylint: disable=W0613
raise NotImplementedError # pragma: no cover
[docs]
@dns.immutable.immutable
class PortParam(Param):
def __init__(self, port):
self.port = dns.rdata.Rdata._as_uint16(port)
@classmethod
def from_value(cls, value):
if not dns._text_util.is_ascii_digits(value):
raise dns.exception.SyntaxError("port is not an integer")
value = int(value)
return cls(value)
def to_text(self):
return f'"{self.port}"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
port = parser.get_uint16()
return cls(port)
def to_wire(self, file, origin=None): # pylint: disable=W0613
file.write(struct.pack("!H", self.port))
[docs]
@dns.immutable.immutable
class IPv4HintParam(Param):
def __init__(self, addresses):
self.addresses = dns.rdata.Rdata._as_tuple(
addresses, dns.rdata.Rdata._as_ipv4_address
)
if len(self.addresses) == 0:
raise ValueError("empty address list")
@classmethod
def from_value(cls, value):
addresses = value.split(",")
return cls(addresses)
def to_text(self):
return '"' + ",".join(self.addresses) + '"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
addresses = []
while parser.remaining() > 0:
ip = parser.get_bytes(4)
addresses.append(dns.ipv4.inet_ntoa(ip))
return cls(addresses)
def to_wire(self, file, origin=None): # pylint: disable=W0613
for address in self.addresses:
file.write(dns.ipv4.inet_aton(address))
[docs]
@dns.immutable.immutable
class IPv6HintParam(Param):
def __init__(self, addresses):
self.addresses = dns.rdata.Rdata._as_tuple(
addresses, dns.rdata.Rdata._as_ipv6_address
)
if len(self.addresses) == 0:
raise ValueError("empty address list")
@classmethod
def from_value(cls, value):
addresses = value.split(",")
return cls(addresses)
def to_text(self):
return '"' + ",".join(self.addresses) + '"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
addresses = []
while parser.remaining() > 0:
ip = parser.get_bytes(16)
addresses.append(dns.ipv6.inet_ntoa(ip))
return cls(addresses)
def to_wire(self, file, origin=None): # pylint: disable=W0613
for address in self.addresses:
file.write(dns.ipv6.inet_aton(address))
[docs]
@dns.immutable.immutable
class ECHParam(Param):
def __init__(self, ech):
self.ech = dns.rdata.Rdata._as_bytes(ech, True)
# An ECHConfigList: a length prefix and 4 or more octets (RFC 9849
# Section 4).
if (
len(self.ech) < 6
or int.from_bytes(self.ech[:2], "big") != len(self.ech) - 2
):
raise ValueError("invalid ECHConfigList")
@classmethod
def from_value(cls, value):
if "\\" in value:
raise ValueError("escape in ECH value")
value = base64.b64decode(value.encode(), validate=True)
return cls(value)
def to_text(self):
b64 = base64.b64encode(self.ech).decode("ascii")
return f'"{b64}"'
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
value = parser.get_bytes(parser.remaining())
return cls(value)
def to_wire(self, file, origin=None): # pylint: disable=W0613
file.write(self.ech)
@dns.immutable.immutable
class OHTTPParam(Param):
# We don't ever expect to instantiate this class, but we need
# a from_value() and a from_wire_parser(), so we just return None
# from the class methods when things are OK.
@classmethod
def emptiness(cls):
return Emptiness.ALWAYS
@classmethod
def from_value(cls, value):
if value is None or value == "":
return None
else:
raise ValueError("ohttp with non-empty value")
def to_text(self):
raise NotImplementedError # pragma: no cover
@classmethod
def from_wire_parser(cls, parser, origin=None): # pylint: disable=W0613
if parser.remaining() != 0:
raise dns.exception.FormError
return None
def to_wire(self, file, origin=None): # pylint: disable=W0613
raise NotImplementedError # pragma: no cover
@dns.immutable.immutable
class DoCPathParam(_StringList):
"""The docpath parameter."""
_class_for_key: dict[ParamKey, Any] = {
ParamKey.MANDATORY: MandatoryParam,
ParamKey.ALPN: ALPNParam,
ParamKey.NO_DEFAULT_ALPN: NoDefaultALPNParam,
ParamKey.PORT: PortParam,
ParamKey.IPV4HINT: IPv4HintParam,
ParamKey.ECH: ECHParam,
ParamKey.IPV6HINT: IPv6HintParam,
ParamKey.OHTTP: OHTTPParam,
ParamKey.DOCPATH: DoCPathParam,
}
def _validate_and_define(params, key, value):
# Keys cannot contain escapes (RFC 9460 Section 2.1).
if "\\" in key:
raise SyntaxError("escape in key")
_check_key_text(key)
key, force_generic = _validate_key(key)
if key in params:
raise SyntaxError(f'duplicate key "{key:d}"')
cls = _class_for_key.get(key, GenericParam)
# An omitted value is the same as an empty one (RFC 9460 Section 2.1).
if (value is None or value == "") and cls.emptiness() == Emptiness.NEVER:
raise SyntaxError("value cannot be empty")
if value is None:
value = cls.from_value(value)
else:
if force_generic:
value = cls.from_wire_parser(dns.wire.Parser(_unescape(value)))
else:
value = cls.from_value(value)
params[key] = value
T = TypeVar("T", bound="SVCBBase")
@dns.immutable.immutable
class SVCBBase(dns.rdata.Rdata):
"""Base class for SVCB-like records"""
# see: draft-ietf-dnsop-svcb-https-11
__slots__ = ["priority", "target", "params"]
def __init__(self, rdclass, rdtype, priority, target, params):
super().__init__(rdclass, rdtype)
self.priority: int = self._as_uint16(priority)
self.target: dns.name.Name = self._as_name(target)
for k, v in params.items():
k = ParamKey.make(k)
if not isinstance(v, Param) and v is not None:
raise ValueError(f"{k:d} not a Param")
if v is None:
pcls = _class_for_key.get(k, GenericParam)
if pcls.emptiness() == Emptiness.NEVER:
raise ValueError(f"key {k:d} cannot be empty")
self.params: dns.immutable.Dict = dns.immutable.Dict(params)
# Make sure any parameter listed as mandatory is present in the
# record.
mandatory = params.get(ParamKey.MANDATORY)
if mandatory is not None:
for key in mandatory.keys:
# Note we have to say "not in" as we have None as a value
# so a get() and a not None test would be wrong.
if key not in params:
raise ValueError(f"key {key:d} declared mandatory but not present")
# The no-default-alpn parameter requires the alpn parameter.
if ParamKey.NO_DEFAULT_ALPN in params:
if ParamKey.ALPN not in params:
raise ValueError("no-default-alpn present, but alpn missing")
def to_styled_text(self, style: dns.rdata.RdataStyle) -> str:
target = self.target.to_styled_text(style)
params = []
for key in sorted(self.params.keys()):
value = self.params[key]
if value is None:
params.append(key_to_text(key))
else:
kv = key_to_text(key) + "=" + value.to_text()
params.append(kv)
if len(params) > 0:
space = " "
else:
space = ""
return f"{self.priority} {target}{space}{' '.join(params)}"
@classmethod
def from_text(
cls: type[T],
rdclass,
rdtype,
tok,
origin=None,
relativize=True,
relativize_to=None,
) -> T:
priority = tok.get_uint16()
target = tok.get_name(origin, relativize, relativize_to)
if priority == 0:
token = tok.get()
if not token.is_eol_or_eof():
raise SyntaxError("parameters in AliasMode")
tok.unget(token)
params = {}
while True:
token = tok.get()
if token.is_eol_or_eof():
tok.unget(token)
break
if token.ttype != dns.tokenizer.IDENTIFIER:
raise SyntaxError("parameter is not an identifier")
equals = token.value.find("=")
if equals == len(token.value) - 1:
# 'key=', so next token should be a quoted string without
# any intervening whitespace.
key = token.value[:-1]
token = tok.get(want_leading=True)
if token.ttype != dns.tokenizer.QUOTED_STRING:
raise SyntaxError("whitespace after =")
value = token.value
elif equals > 0:
# key=value
key = token.value[:equals]
value = token.value[equals + 1 :]
elif equals == 0:
# =key
raise SyntaxError('parameter cannot start with "="')
else:
# key
key = token.value
value = None
_validate_and_define(params, key, value)
return cls(rdclass, rdtype, priority, target, params)
def _to_wire(self, file, compress=None, origin=None, canonicalize=False):
file.write(struct.pack("!H", self.priority))
self.target.to_wire(file, None, origin, False)
for key in sorted(self.params):
file.write(struct.pack("!H", key))
value = self.params[key]
with dns.renderer.prefixed_length(file, 2):
# Note that we're still writing a length of zero if the value is None
if value is not None:
value.to_wire(file, origin)
@classmethod
def from_wire_parser(cls: type[T], rdclass, rdtype, parser, origin=None) -> T:
priority = parser.get_uint16()
target = parser.get_name(origin)
if priority == 0 and parser.remaining() != 0:
raise dns.exception.FormError("parameters in AliasMode")
params = {}
prior_key = -1
while parser.remaining() > 0:
key = parser.get_uint16()
if key <= prior_key:
raise dns.exception.FormError("keys not in strictly increasing order")
prior_key = key
vlen = parser.get_uint16()
pkey = ParamKey.make(key)
pcls = _class_for_key.get(pkey, GenericParam)
if vlen == 0 and pcls.emptiness() == Emptiness.NEVER:
raise dns.exception.FormError(f"key {key:d} cannot be empty")
with parser.restrict_to(vlen):
value = pcls.from_wire_parser(parser, origin)
params[pkey] = value
return cls(rdclass, rdtype, priority, target, params)
def _processing_priority(self):
return self.priority
@classmethod
def _processing_order(cls, iterable):
return dns.rdtypes.util.priority_processing_order(iterable)