1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
# 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): # noqa: B007
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
|
