# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
# Copyright (C) 2006, 2007, 2009-2011 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.
import collections
import random
import struct
from typing import Any, List
import dns.exception
import dns.ipv4
import dns.ipv6
import dns.name
import dns.rdata
class Gateway:
"""A helper class for the IPSECKEY gateway and AMTRELAY relay fields"""
name = ""
def __init__(self, type, gateway=None):
self.type = dns.rdata.Rdata._as_uint8(type)
self.gateway = gateway
self._check()
@classmethod
def _invalid_type(cls, gateway_type):
return f"invalid {cls.name} type: {gateway_type}"
def _check(self):
if self.type == 0:
if self.gateway not in (".", None):
raise SyntaxError(f"invalid {self.name} for type 0")
self.gateway = None
elif self.type == 1:
# check that it's OK
dns.ipv4.inet_aton(self.gateway)
elif self.type == 2:
# check that it's OK
dns.ipv6.inet_aton(self.gateway)
elif self.type == 3:
if not isinstance(self.gateway, dns.name.Name):
raise SyntaxError(f"invalid {self.name}; not a name")
else:
raise SyntaxError(self._invalid_type(self.type))
def to_text(self, origin=None, relativize=True):
if self.type == 0:
return "."
elif self.type in (1, 2):
return self.gateway
elif self.type == 3:
return str(self.gateway.choose_relativity(origin, relativize))
else:
raise ValueError(self._invalid_type(self.type)) # pragma: no cover
@classmethod
def from_text(
cls, gateway_type, tok, origin=None, relativize=True, relativize_to=None
):
if gateway_type in (0, 1, 2):
gateway = tok.get_string()
elif gateway_type == 3:
gateway = tok.get_name(origin, relativize, relativize_to)
else:
raise dns.exception.SyntaxError(
cls._invalid_type(gateway_type)
) # pragma: no cover
return cls(gateway_type, gateway)
# pylint: disable=unused-argument
def to_wire(self, file, compress=None, origin=None, canonicalize=False):
if self.type == 0:
pass
elif self.type == 1:
file.write(dns.ipv4.inet_aton(self.gateway))
elif self.type == 2:
file.write(dns.ipv6.inet_aton(self.gateway))
elif self.type == 3:
self.gateway.to_wire(file, None, origin, False)
else:
raise ValueError(self._invalid_type(self.type)) # pragma: no cover
# pylint: enable=unused-argument
@classmethod
def from_wire_parser(cls, gateway_type, parser, origin=None):
if gateway_type == 0:
gateway = None
elif gateway_type == 1:
gateway = dns.ipv4.inet_ntoa(parser.get_bytes(4))
elif gateway_type == 2:
gateway = dns.ipv6.inet_ntoa(parser.get_bytes(16))
elif gateway_type == 3:
gateway = parser.get_name(origin)
else:
raise dns.exception.FormError(cls._invalid_type(gateway_type))
return cls(gateway_type, gateway)
class Bitmap:
"""A helper class for the NSEC/NSEC3/CSYNC type bitmaps"""
type_name = ""
def __init__(self, windows=None):
last_window = -1
self.windows = windows
for window, bitmap in self.windows:
if not isinstance(window, int):
raise ValueError(f"bad {self.type_name} window type")
if window <= last_window:
raise ValueError(f"bad {self.type_name} window order")
if window > 256:
raise ValueError(f"bad {self.type_name} window number")
last_window = window
if not isinstance(bitmap, bytes):
raise ValueError(f"bad {self.type_name} octets type")
if len(bitmap) == 0 or len(bitmap) > 32:
raise ValueError(f"bad {self.type_name} octets")
def to_text(self) -> str:
text = ""
for window, bitmap in self.windows:
bits = []
for i, byte in enumerate(bitmap):
for j in range(0, 8):
if byte & (0x80 >> j):
rdtype = window * 256 + i * 8 + j
bits.append(dns.rdatatype.to_text(rdtype))
text += " " + " ".join(bits)
return text
@classmethod
def from_text(cls, tok: "dns.tokenizer.Tokenizer") -> "Bitmap":
rdtypes = []
for token in tok.get_remaining():
rdtype = dns.rdatatype.from_text(token.unescape().value)
if rdtype == 0:
raise dns.exception.SyntaxError(f"{cls.type_name} with bit 0")
rdtypes.append(rdtype)
return cls.from_rdtypes(rdtypes)
@classmethod
def from_rdtypes(cls, rdtypes: List[dns.rdatatype.RdataType]) -> "Bitmap":
rdtypes = sorted(rdtypes)
window = 0
octets = 0
prior_rdtype = 0
bitmap = bytearray(b"\0" * 32)
windows = []
for rdtype in rdtypes:
if rdtype == prior_rdtype:
continue
prior_rdtype = rdtype
new_window = rdtype // 256
if new_window != window:
if octets != 0:
windows.append((window, bytes(bitmap[0:octets])))
bitmap = bytearray(b"\0" * 32)
window = new_window
offset = rdtype % 256
byte = offset // 8
bit = offset % 8
octets = byte + 1
bitmap[byte] = bitmap[byte] | (0x80 >> bit)
if octets != 0:
windows.append((window, bytes(bitmap[0:octets])))
return cls(windows)
def to_wire(self, file: Any) -> None:
for window, bitmap in self.windows:
file.write(struct.pack("!BB", window, len(bitmap)))
file.write(bitmap)
@classmethod
def from_wire_parser(cls, parser: "dns.wire.Parser") -> "Bitmap":
windows = []
while parser.remaining() > 0:
window = parser.get_uint8()
bitmap = parser.get_counted_bytes()
windows.append((window, bitmap))
return cls(windows)
def _priority_table(items):
by_priority = collections.defaultdict(list)
for rdata in items:
by_priority[rdata._processing_priority()].append(rdata)
return by_priority
def priority_processing_order(iterable):
items = list(iterable)
if len(items) == 1:
return items
by_priority = _priority_table(items)
ordered = []
for k in sorted(by_priority.keys()):
rdatas = by_priority[k]
random.shuffle(rdatas)
ordered.extend(rdatas)
return ordered
_no_weight = 0.1
def weighted_processing_order(iterable):
items = list(iterable)
if len(items) == 1:
return items
by_priority = _priority_table(items)
ordered = []
for k in sorted(by_priority.keys()):
rdatas = by_priority[k]
total = sum(rdata._processing_weight() or _no_weight for rdata in rdatas)
while len(rdatas) > 1:
r = random.uniform(0, total)
for n, rdata in enumerate(rdatas):
weight = rdata._processing_weight() or _no_weight
if weight > r:
break
r -= weight
total -= weight
ordered.append(rdata) # pylint: disable=undefined-loop-variable
del rdatas[n] # pylint: disable=undefined-loop-variable
ordered.append(rdatas[0])
return ordered
def parse_formatted_hex(formatted, num_chunks, chunk_size, separator):
if len(formatted) != num_chunks * (chunk_size + 1) - 1:
raise ValueError("invalid formatted hex string")
value = b""
for _ in range(num_chunks):
chunk = formatted[0:chunk_size]
value += int(chunk, 16).to_bytes(chunk_size // 2, "big")
formatted = formatted[chunk_size:]
if len(formatted) > 0 and formatted[0] != separator:
raise ValueError("invalid formatted hex string")
formatted = formatted[1:]
return value
