Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/dns.py: 45%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) Philippe Biondi <phil@secdev.org>
6"""
7DNS: Domain Name System
9This implements:
10- RFC1035: Domain Names
11- RFC6762: Multicast DNS
12- RFC6763: DNS-Based Service Discovery
13"""
15import abc
16import operator
17import itertools
18import socket
19import struct
20import time
21import warnings
23from scapy.arch import (
24 get_if_addr,
25 get_if_addr6,
26 read_nameservers,
27)
28from scapy.ansmachine import AnsweringMachine
29from scapy.base_classes import Net, ScopedIP
30from scapy.config import conf
31from scapy.compat import raw, chb, bytes_encode, plain_str
32from scapy.error import log_runtime, warning, Scapy_Exception
33from scapy.packet import Packet, bind_layers, Raw
34from scapy.fields import (
35 BitEnumField,
36 BitField,
37 ByteEnumField,
38 ByteField,
39 ConditionalField,
40 Field,
41 FieldLenField,
42 FieldListField,
43 FlagsField,
44 I,
45 IP6Field,
46 IntField,
47 MACField,
48 MultipleTypeField,
49 PacketListField,
50 ShortEnumField,
51 ShortField,
52 StrField,
53 StrLenField,
54 UTCTimeField,
55 XStrFixedLenField,
56 XStrLenField,
57)
58from scapy.interfaces import resolve_iface
59from scapy.sendrecv import sr1, sr
60from scapy.supersocket import StreamSocket
61from scapy.plist import SndRcvList, _PacketList, QueryAnswer
62from scapy.pton_ntop import inet_ntop, inet_pton
63from scapy.utils import pretty_list
64from scapy.volatile import RandShort
66from scapy.layers.l2 import Ether
67from scapy.layers.inet import IP, DestIPField, IPField, UDP, TCP
68from scapy.layers.inet6 import IPv6
70from typing import (
71 Any,
72 List,
73 Optional,
74 Tuple,
75 Type,
76 Union,
77)
80# https://www.iana.org/assignments/dns-parameters/dns-parameters.xhtml#dns-parameters-4
81dnstypes = {
82 0: "RESERVED",
83 1: "A", 2: "NS", 3: "MD", 4: "MF", 5: "CNAME", 6: "SOA", 7: "MB", 8: "MG",
84 9: "MR", 10: "NULL", 11: "WKS", 12: "PTR", 13: "HINFO", 14: "MINFO",
85 15: "MX", 16: "TXT", 17: "RP", 18: "AFSDB", 19: "X25", 20: "ISDN",
86 21: "RT", 22: "NSAP", 23: "NSAP-PTR", 24: "SIG", 25: "KEY", 26: "PX",
87 27: "GPOS", 28: "AAAA", 29: "LOC", 30: "NXT", 31: "EID", 32: "NIMLOC",
88 33: "SRV", 34: "ATMA", 35: "NAPTR", 36: "KX", 37: "CERT", 38: "A6",
89 39: "DNAME", 40: "SINK", 41: "OPT", 42: "APL", 43: "DS", 44: "SSHFP",
90 45: "IPSECKEY", 46: "RRSIG", 47: "NSEC", 48: "DNSKEY", 49: "DHCID",
91 50: "NSEC3", 51: "NSEC3PARAM", 52: "TLSA", 53: "SMIMEA", 55: "HIP",
92 56: "NINFO", 57: "RKEY", 58: "TALINK", 59: "CDS", 60: "CDNSKEY",
93 61: "OPENPGPKEY", 62: "CSYNC", 63: "ZONEMD", 64: "SVCB", 65: "HTTPS",
94 99: "SPF", 100: "UINFO", 101: "UID", 102: "GID", 103: "UNSPEC", 104: "NID",
95 105: "L32", 106: "L64", 107: "LP", 108: "EUI48", 109: "EUI64", 249: "TKEY",
96 250: "TSIG", 256: "URI", 257: "CAA", 258: "AVC", 259: "DOA",
97 260: "AMTRELAY", 32768: "TA", 32769: "DLV", 65535: "RESERVED"
98}
101dnsqtypes = {251: "IXFR", 252: "AXFR", 253: "MAILB", 254: "MAILA", 255: "ALL"}
102dnsqtypes.update(dnstypes)
103dnsclasses = {1: 'IN', 2: 'CS', 3: 'CH', 4: 'HS', 255: 'ANY'}
106# 12/2023 from https://www.iana.org/assignments/dns-sec-alg-numbers/dns-sec-alg-numbers.xhtml # noqa: E501
107dnssecalgotypes = {0: "Reserved", 1: "RSA/MD5", 2: "Diffie-Hellman", 3: "DSA/SHA-1", # noqa: E501
108 4: "Reserved", 5: "RSA/SHA-1", 6: "DSA-NSEC3-SHA1",
109 7: "RSASHA1-NSEC3-SHA1", 8: "RSA/SHA-256", 9: "Reserved",
110 10: "RSA/SHA-512", 11: "Reserved", 12: "GOST R 34.10-2001",
111 13: "ECDSA Curve P-256 with SHA-256", 14: "ECDSA Curve P-384 with SHA-384", # noqa: E501
112 15: "Ed25519", 16: "Ed448",
113 252: "Reserved for Indirect Keys", 253: "Private algorithms - domain name", # noqa: E501
114 254: "Private algorithms - OID", 255: "Reserved"}
116# 12/2023 from https://www.iana.org/assignments/ds-rr-types/ds-rr-types.xhtml
117dnssecdigesttypes = {0: "Reserved", 1: "SHA-1", 2: "SHA-256", 3: "GOST R 34.11-94", 4: "SHA-384"} # noqa: E501
119# 12/2023 from https://www.iana.org/assignments/dnssec-nsec3-parameters/dnssec-nsec3-parameters.xhtml # noqa: E501
120dnssecnsec3algotypes = {0: "Reserved", 1: "SHA-1"}
123def dns_get_str(s, full=None, _ignore_compression=False):
124 """This function decompresses a string s, starting
125 from the given pointer.
127 :param s: the string to decompress
128 :param full: (optional) the full packet (used for decompression)
130 :returns: (decoded_string, end_index, left_string)
131 """
132 # _ignore_compression is for internal use only
133 max_length = len(s)
134 # The result = the extracted name
135 name = b""
136 # Will contain the index after the pointer, to be returned
137 after_pointer = None
138 processed_pointers = [] # Used to check for decompression loops
139 bytes_left = None
140 _fullpacket = False # s = full packet
141 pointer = 0
142 while True:
143 if abs(pointer) >= max_length:
144 log_runtime.info(
145 "DNS RR prematured end (ofs=%i, len=%i)", pointer, len(s)
146 )
147 break
148 cur = s[pointer] # get pointer value
149 pointer += 1 # make pointer go forward
150 if cur & 0xc0: # Label pointer
151 if after_pointer is None:
152 # after_pointer points to where the remaining bytes start,
153 # as pointer will follow the jump token
154 after_pointer = pointer + 1
155 if _ignore_compression:
156 # skip
157 pointer += 1
158 continue
159 if pointer >= max_length:
160 log_runtime.info(
161 "DNS incomplete jump token at (ofs=%i)", pointer
162 )
163 break
164 if not full:
165 raise Scapy_Exception("DNS message can't be compressed " +
166 "at this point!")
167 # Follow the pointer
168 pointer = ((cur & ~0xc0) << 8) + s[pointer]
169 if pointer in processed_pointers:
170 warning("DNS decompression loop detected")
171 break
172 if len(processed_pointers) >= 20:
173 warning("More than 20 jumps in a single DNS decompression ! "
174 "Dropping (evil packet)")
175 break
176 if not _fullpacket:
177 # We switch our s buffer to full, so we need to remember
178 # the previous context
179 bytes_left = s[after_pointer:]
180 s = full
181 max_length = len(s)
182 _fullpacket = True
183 processed_pointers.append(pointer)
184 continue
185 elif cur > 0: # Label
186 # cur = length of the string
187 name += s[pointer:pointer + cur] + b"."
188 pointer += cur
189 else: # End
190 break
191 if after_pointer is not None:
192 # Return the real end index (not the one we followed)
193 pointer = after_pointer
194 if bytes_left is None:
195 bytes_left = s[pointer:]
196 # name, remaining
197 return name or b".", bytes_left
200def _is_ptr(x):
201 """
202 Heuristic to guess if bytes are an encoded DNS pointer.
203 """
204 return (
205 (x and x[-1] == 0) or
206 (len(x) >= 2 and (x[-2] & 0xc0) == 0xc0)
207 )
210def dns_encode(x, check_built=False):
211 """Encodes a bytes string into the DNS format
213 :param x: the string
214 :param check_built: detect already-built strings and ignore them
215 :returns: the encoded bytes string
216 """
217 if not x or x == b".":
218 return b"\x00"
220 if check_built and _is_ptr(x):
221 # The value has already been processed. Do not process it again
222 return x
224 # Truncate chunks that cannot be encoded (more than 63 bytes..)
225 x = b"".join(chb(len(y)) + y for y in (k[:63] for k in x.split(b".")))
226 if x[-1:] != b"\x00":
227 x += b"\x00"
228 return x
231def DNSgetstr(*args, **kwargs):
232 """Legacy function. Deprecated"""
233 warnings.warn(
234 "DNSgetstr is deprecated. Use dns_get_str instead.",
235 DeprecationWarning
236 )
237 return dns_get_str(*args, **kwargs)[:-1]
240def dns_compress(pkt):
241 """This function compresses a DNS packet according to compression rules.
242 """
243 if DNS not in pkt:
244 raise Scapy_Exception("Can only compress DNS layers")
245 pkt = pkt.copy()
246 dns_pkt = pkt.getlayer(DNS)
247 dns_pkt.clear_cache()
248 build_pkt = raw(dns_pkt)
250 def field_gen(dns_pkt):
251 """Iterates through all DNS strings that can be compressed"""
252 for lay in [dns_pkt.qd, dns_pkt.an, dns_pkt.ns, dns_pkt.ar]:
253 if not lay:
254 continue
255 for current in lay:
256 for field in current.fields_desc:
257 if isinstance(field, DNSStrField) or \
258 (isinstance(field, MultipleTypeField) and
259 current.type in [2, 3, 4, 5, 12, 15, 39, 47]):
260 # Get the associated data and store it accordingly # noqa: E501
261 dat = current.getfieldval(field.name)
262 yield current, field.name, dat
264 def possible_shortens(dat):
265 """Iterates through all possible compression parts in a DNS string"""
266 if dat == b".": # we'd lose by compressing it
267 return
268 yield dat
269 for x in range(1, dat.count(b".")):
270 yield dat.split(b".", x)[x]
271 data = {}
272 for current, name, dat in field_gen(dns_pkt):
273 for part in possible_shortens(dat):
274 # Encode the data
275 encoded = dns_encode(part, check_built=True)
276 if part not in data:
277 # We have no occurrence of such data, let's store it as a
278 # possible pointer for future strings.
279 # We get the index of the encoded data
280 index = build_pkt.index(encoded)
281 # The following is used to build correctly the pointer
282 fb_index = ((index >> 8) | 0xc0)
283 sb_index = index - (256 * (fb_index - 0xc0))
284 pointer = chb(fb_index) + chb(sb_index)
285 data[part] = [(current, name, pointer, index + 1)]
286 else:
287 # This string already exists, let's mark the current field
288 # with it, so that it gets compressed
289 data[part].append((current, name))
290 _in = data[part][0][3]
291 build_pkt = build_pkt[:_in] + build_pkt[_in:].replace(
292 encoded,
293 b"\0\0",
294 1
295 )
296 break
297 # Apply compression rules
298 for ck in data:
299 # compression_key is a DNS string
300 replacements = data[ck]
301 # replacements is the list of all tuples (layer, field name)
302 # where this string was found
303 replace_pointer = replacements.pop(0)[2]
304 # replace_pointer is the packed pointer that should replace
305 # those strings. Note that pop remove it from the list
306 for rep in replacements:
307 # setfieldval edits the value of the field in the layer
308 val = rep[0].getfieldval(rep[1])
309 assert val.endswith(ck)
310 kept_string = dns_encode(val[:-len(ck)], check_built=True)[:-1]
311 new_val = kept_string + replace_pointer
312 rep[0].setfieldval(rep[1], new_val)
313 try:
314 del rep[0].rdlen
315 except AttributeError:
316 pass
317 # End of the compression algorithm
318 # Destroy the previous DNS layer if needed
319 if not isinstance(pkt, DNS) and pkt.getlayer(DNS).underlayer:
320 pkt.getlayer(DNS).underlayer.remove_payload()
321 return pkt / dns_pkt
322 return dns_pkt
325class DNSCompressedPacket(Packet):
326 """
327 Class to mark that a packet contains DNSStrField and supports compression
328 """
329 @abc.abstractmethod
330 def get_full(self):
331 pass
334class DNSStrField(StrLenField):
335 """
336 Special StrField that handles DNS encoding/decoding.
337 It will also handle DNS decompression.
338 (may be StrLenField if a length_from is passed),
339 """
340 def any2i(self, pkt, x):
341 if x and isinstance(x, list):
342 return [self.h2i(pkt, y) for y in x]
343 return super(DNSStrField, self).any2i(pkt, x)
345 def h2i(self, pkt, x):
346 # Setting a DNSStrField manually (h2i) means any current compression will break
347 if (
348 pkt and
349 isinstance(pkt.parent, DNSCompressedPacket) and
350 pkt.parent.raw_packet_cache
351 ):
352 pkt.parent.clear_cache()
353 if not x:
354 return b"."
355 x = bytes_encode(x)
356 if x[-1:] != b"." and not _is_ptr(x):
357 return x + b"."
358 return x
360 def i2m(self, pkt, x):
361 return dns_encode(x, check_built=True)
363 def i2len(self, pkt, x):
364 return len(self.i2m(pkt, x))
366 def get_full(self, pkt):
367 while pkt and not isinstance(pkt, DNSCompressedPacket):
368 pkt = pkt.parent or pkt.underlayer
369 if not pkt:
370 return None
371 return pkt.get_full()
373 def getfield(self, pkt, s):
374 remain = b""
375 if self.length_from:
376 remain, s = super(DNSStrField, self).getfield(pkt, s)
377 # Decode the compressed DNS message
378 decoded, left = dns_get_str(s, full=self.get_full(pkt))
379 # returns (remaining, decoded)
380 return left + remain, decoded
383class DNSTextField(StrLenField):
384 """
385 Special StrLenField that handles DNS TEXT data (16)
386 """
388 islist = 1
390 def i2h(self, pkt, x):
391 if not x:
392 return []
393 return x
395 def m2i(self, pkt, s):
396 ret_s = list()
397 tmp_s = s
398 # RDATA contains a list of strings, each are prepended with
399 # a byte containing the size of the following string.
400 while tmp_s:
401 tmp_len = tmp_s[0] + 1
402 if tmp_len > len(tmp_s):
403 log_runtime.info(
404 "DNS RR TXT prematured end of character-string "
405 "(size=%i, remaining bytes=%i)", tmp_len, len(tmp_s)
406 )
407 ret_s.append(tmp_s[1:tmp_len])
408 tmp_s = tmp_s[tmp_len:]
409 return ret_s
411 def any2i(self, pkt, x):
412 if isinstance(x, (str, bytes)):
413 return [x]
414 return x
416 def i2len(self, pkt, x):
417 return len(self.i2m(pkt, x))
419 def i2m(self, pkt, s):
420 ret_s = b""
421 for text in s:
422 if not text:
423 ret_s += b"\x00"
424 continue
425 text = bytes_encode(text)
426 # The initial string must be split into a list of strings
427 # prepended with theirs sizes.
428 while len(text) >= 255:
429 ret_s += b"\xff" + text[:255]
430 text = text[255:]
431 # The remaining string is less than 255 bytes long
432 if len(text):
433 ret_s += struct.pack("!B", len(text)) + text
434 return ret_s
437# RFC 2671 - Extension Mechanisms for DNS (EDNS0)
439edns0types = {0: "Reserved", 1: "LLQ", 2: "UL", 3: "NSID", 4: "Owner",
440 5: "DAU", 6: "DHU", 7: "N3U", 8: "edns-client-subnet", 10: "COOKIE",
441 12: "Padding", 15: "Extended DNS Error"}
444class _EDNS0Dummy(Packet):
445 name = "Dummy class that implements extract_padding()"
447 def extract_padding(self, p):
448 # type: (bytes) -> Tuple[bytes, Optional[bytes]]
449 return "", p
452class EDNS0TLV(_EDNS0Dummy):
453 name = "DNS EDNS0 TLV"
454 fields_desc = [ShortEnumField("optcode", 0, edns0types),
455 FieldLenField("optlen", None, "optdata", fmt="H"),
456 StrLenField("optdata", "",
457 length_from=lambda pkt: pkt.optlen)]
459 @classmethod
460 def dispatch_hook(cls, _pkt=None, *args, **kargs):
461 # type: (Optional[bytes], *Any, **Any) -> Type[Packet]
462 if _pkt is None:
463 return EDNS0TLV
464 if len(_pkt) < 2:
465 return Raw
466 edns0type = struct.unpack("!H", _pkt[:2])[0]
467 return EDNS0OPT_DISPATCHER.get(edns0type, EDNS0TLV)
470class DNSRROPT(Packet):
471 name = "DNS OPT Resource Record"
472 fields_desc = [DNSStrField("rrname", ""),
473 ShortEnumField("type", 41, dnstypes),
474 ShortEnumField("rclass", 4096, dnsclasses),
475 ByteField("extrcode", 0),
476 ByteField("version", 0),
477 # version 0 means EDNS0
478 BitEnumField("z", 32768, 16, {32768: "D0"}),
479 # D0 means DNSSEC OK from RFC 3225
480 FieldLenField("rdlen", None, length_of="rdata", fmt="H"),
481 PacketListField("rdata", [], EDNS0TLV,
482 length_from=lambda pkt: pkt.rdlen)]
485# draft-cheshire-edns0-owner-option-01 - EDNS0 OWNER Option
487class EDNS0OWN(_EDNS0Dummy):
488 name = "EDNS0 Owner (OWN)"
489 fields_desc = [ShortEnumField("optcode", 4, edns0types),
490 FieldLenField("optlen", None, count_of="primary_mac", fmt="H"),
491 ByteField("v", 0),
492 ByteField("s", 0),
493 MACField("primary_mac", "00:00:00:00:00:00"),
494 ConditionalField(
495 MACField("wakeup_mac", "00:00:00:00:00:00"),
496 lambda pkt: (pkt.optlen or 0) >= 14),
497 ConditionalField(
498 StrLenField("password", "",
499 length_from=lambda pkt: pkt.optlen - 14),
500 lambda pkt: (pkt.optlen or 0) >= 18)]
502 def post_build(self, pkt, pay):
503 pkt += pay
504 if self.optlen is None:
505 pkt = pkt[:2] + struct.pack("!H", len(pkt) - 4) + pkt[4:]
506 return pkt
509# RFC 6975 - Signaling Cryptographic Algorithm Understanding in
510# DNS Security Extensions (DNSSEC)
512class EDNS0DAU(_EDNS0Dummy):
513 name = "DNSSEC Algorithm Understood (DAU)"
514 fields_desc = [ShortEnumField("optcode", 5, edns0types),
515 FieldLenField("optlen", None, count_of="alg_code", fmt="H"),
516 FieldListField("alg_code", None,
517 ByteEnumField("", 0, dnssecalgotypes),
518 count_from=lambda pkt:pkt.optlen)]
521class EDNS0DHU(_EDNS0Dummy):
522 name = "DS Hash Understood (DHU)"
523 fields_desc = [ShortEnumField("optcode", 6, edns0types),
524 FieldLenField("optlen", None, count_of="alg_code", fmt="H"),
525 FieldListField("alg_code", None,
526 ByteEnumField("", 0, dnssecdigesttypes),
527 count_from=lambda pkt:pkt.optlen)]
530class EDNS0N3U(_EDNS0Dummy):
531 name = "NSEC3 Hash Understood (N3U)"
532 fields_desc = [ShortEnumField("optcode", 7, edns0types),
533 FieldLenField("optlen", None, count_of="alg_code", fmt="H"),
534 FieldListField("alg_code", None,
535 ByteEnumField("", 0, dnssecnsec3algotypes),
536 count_from=lambda pkt:pkt.optlen)]
539# RFC 7871 - Client Subnet in DNS Queries
541class ClientSubnetv4(StrLenField):
542 af_familly = socket.AF_INET
543 af_length = 32
544 af_default = b"\xc0" # 192.0.0.0
546 def getfield(self, pkt, s):
547 # type: (Packet, bytes) -> Tuple[bytes, I]
548 sz = operator.floordiv(self.length_from(pkt) + 7, 8)
549 sz = min(sz, operator.floordiv(self.af_length, 8))
550 return s[sz:], self.m2i(pkt, s[:sz])
552 def m2i(self, pkt, x):
553 # type: (Optional[Packet], bytes) -> str
554 padding = self.af_length - self.length_from(pkt)
555 if padding:
556 x += b"\x00" * operator.floordiv(padding, 8)
557 x = x[: operator.floordiv(self.af_length, 8)]
558 return inet_ntop(self.af_familly, x)
560 def _pack_subnet(self, subnet, plen=None):
561 # type: (bytes, Optional[int]) -> bytes
562 packed_subnet = inet_pton(self.af_familly, plain_str(subnet))
563 if plen is not None:
564 # RFC 7871: ADDRESS MUST be truncated to the number of bits
565 # indicated by SOURCE PREFIX-LENGTH, padded with 0 bits to the
566 # end of the last octet needed. Use ceil(plen / 8) bytes and
567 # zero out any host bits in the last partial byte.
568 num_bytes = operator.floordiv(plen + 7, 8)
569 result = bytearray(packed_subnet[:num_bytes])
570 rem = plen % 8
571 if rem and result:
572 result[-1] &= (0xff << (8 - rem)) & 0xff
573 return bytes(result)
574 # When prefix length is not known, strip trailing zero bytes.
575 for i in list(range(operator.floordiv(self.af_length, 8)))[::-1]:
576 if packed_subnet[i] != 0:
577 i += 1
578 break
579 return packed_subnet[:i]
581 def i2m(self, pkt, x):
582 # type: (Optional[Packet], Optional[Union[str, Net]]) -> bytes
583 if x is None:
584 return self.af_default
585 plen = getattr(pkt, 'source_plen', None)
586 try:
587 return self._pack_subnet(x, plen)
588 except (OSError, socket.error):
589 pkt.family = 2
590 return ClientSubnetv6("", "")._pack_subnet(x, plen)
592 def i2len(self, pkt, x):
593 # type: (Packet, Any) -> int
594 if x is None:
595 return 1
596 plen = getattr(pkt, 'source_plen', None)
597 try:
598 return len(self._pack_subnet(x, plen))
599 except (OSError, socket.error):
600 pkt.family = 2
601 return len(ClientSubnetv6("", "")._pack_subnet(x, plen))
604class ClientSubnetv6(ClientSubnetv4):
605 af_familly = socket.AF_INET6
606 af_length = 128
607 af_default = b"\x20" # 2000::
610class EDNS0ClientSubnet(_EDNS0Dummy):
611 name = "DNS EDNS0 Client Subnet"
612 fields_desc = [ShortEnumField("optcode", 8, edns0types),
613 FieldLenField("optlen", None, "address", fmt="H",
614 adjust=lambda pkt, x: x + 4),
615 ShortField("family", 1),
616 FieldLenField("source_plen", None,
617 length_of="address",
618 fmt="B",
619 adjust=lambda pkt, x: x * 8),
620 ByteField("scope_plen", 0),
621 MultipleTypeField(
622 [(ClientSubnetv4("address", "192.168.0.0",
623 length_from=lambda p: p.source_plen),
624 lambda pkt: pkt.family == 1),
625 (ClientSubnetv6("address", "2001:db8::",
626 length_from=lambda p: p.source_plen),
627 lambda pkt: pkt.family == 2)],
628 ClientSubnetv4("address", "192.168.0.0",
629 length_from=lambda p: p.source_plen))]
632class EDNS0COOKIE(_EDNS0Dummy):
633 name = "DNS EDNS0 COOKIE"
634 fields_desc = [ShortEnumField("optcode", 10, edns0types),
635 FieldLenField("optlen", None, length_of="server_cookie", fmt="!H",
636 adjust=lambda pkt, x: x + 8),
637 XStrFixedLenField("client_cookie", b"\x00" * 8, length=8),
638 XStrLenField("server_cookie", "",
639 length_from=lambda pkt: max(0, pkt.optlen - 8))]
642# RFC 7830 - EDNS0 Padding Option
644class EDNS0PADDING(_EDNS0Dummy):
645 name = "DNS EDNS0 Padding"
646 fields_desc = [ShortEnumField("optcode", 12, edns0types),
647 FieldLenField("optlen", None, length_of="padding", fmt="!H"),
648 StrLenField("padding", "",
649 length_from=lambda pkt: pkt.optlen)]
652# RFC 8914 - Extended DNS Errors
654# https://www.iana.org/assignments/dns-parameters/dns-parameters.xhtml#extended-dns-error-codes
655extended_dns_error_codes = {
656 0: "Other",
657 1: "Unsupported DNSKEY Algorithm",
658 2: "Unsupported DS Digest Type",
659 3: "Stale Answer",
660 4: "Forged Answer",
661 5: "DNSSEC Indeterminate",
662 6: "DNSSEC Bogus",
663 7: "Signature Expired",
664 8: "Signature Not Yet Valid",
665 9: "DNSKEY Missing",
666 10: "RRSIGs Missing",
667 11: "No Zone Key Bit Set",
668 12: "NSEC Missing",
669 13: "Cached Error",
670 14: "Not Ready",
671 15: "Blocked",
672 16: "Censored",
673 17: "Filtered",
674 18: "Prohibited",
675 19: "Stale NXDOMAIN Answer",
676 20: "Not Authoritative",
677 21: "Not Supported",
678 22: "No Reachable Authority",
679 23: "Network Error",
680 24: "Invalid Data",
681 25: "Signature Expired before Valid",
682 26: "Too Early",
683 27: "Unsupported NSEC3 Iterations Value",
684 28: "Unable to conform to policy",
685 29: "Synthesized",
686}
689# https://www.rfc-editor.org/rfc/rfc8914.html
690class EDNS0ExtendedDNSError(_EDNS0Dummy):
691 name = "DNS EDNS0 Extended DNS Error"
692 fields_desc = [ShortEnumField("optcode", 15, edns0types),
693 FieldLenField("optlen", None, length_of="extra_text", fmt="!H",
694 adjust=lambda pkt, x: x + 2),
695 ShortEnumField("info_code", 0, extended_dns_error_codes),
696 StrLenField("extra_text", "",
697 length_from=lambda pkt: pkt.optlen - 2)]
700EDNS0OPT_DISPATCHER = {
701 4: EDNS0OWN,
702 5: EDNS0DAU,
703 6: EDNS0DHU,
704 7: EDNS0N3U,
705 8: EDNS0ClientSubnet,
706 10: EDNS0COOKIE,
707 12: EDNS0PADDING,
708 15: EDNS0ExtendedDNSError,
709}
712# RFC 4034 - Resource Records for the DNS Security Extensions
714def bitmap2RRlist(bitmap):
715 """
716 Decode the 'Type Bit Maps' field of the NSEC Resource Record into an
717 integer list.
718 """
719 # RFC 4034, 4.1.2. The Type Bit Maps Field
721 RRlist = []
723 while bitmap:
725 if len(bitmap) < 2:
726 log_runtime.info("bitmap too short (%i)", len(bitmap))
727 return
729 window_block = bitmap[0] # window number
730 offset = 256 * window_block # offset of the Resource Record
731 bitmap_len = bitmap[1] # length of the bitmap in bytes
733 if bitmap_len <= 0 or bitmap_len > 32:
734 log_runtime.info("bitmap length is no valid (%i)", bitmap_len)
735 return
737 tmp_bitmap = bitmap[2:2 + bitmap_len]
739 # Let's compare each bit of tmp_bitmap and compute the real RR value
740 for b in range(len(tmp_bitmap)):
741 v = 128
742 for i in range(8):
743 if tmp_bitmap[b] & v:
744 # each of the RR is encoded as a bit
745 RRlist += [offset + b * 8 + i]
746 v = v >> 1
748 # Next block if any
749 bitmap = bitmap[2 + bitmap_len:]
751 return RRlist
754def RRlist2bitmap(lst):
755 """
756 Encode a list of integers representing Resource Records to a bitmap field
757 used in the NSEC Resource Record.
758 """
759 # RFC 4034, 4.1.2. The Type Bit Maps Field
761 import math
763 bitmap = b""
764 lst = [abs(x) for x in sorted(set(lst)) if x <= 65535]
766 # number of window blocks
767 max_window_blocks = int(math.ceil(lst[-1] / 256.))
768 min_window_blocks = int(math.floor(lst[0] / 256.))
769 if min_window_blocks == max_window_blocks:
770 max_window_blocks += 1
772 for wb in range(min_window_blocks, max_window_blocks + 1):
773 # First, filter out RR not encoded in the current window block
774 # i.e. keep everything between 256*wb <= 256*(wb+1)
775 rrlist = sorted(x for x in lst if 256 * wb <= x < 256 * (wb + 1))
776 if not rrlist:
777 continue
779 # Compute the number of bytes used to store the bitmap
780 if rrlist[-1] == 0: # only one element in the list
781 bytes_count = 1
782 else:
783 max = rrlist[-1] - 256 * wb
784 bytes_count = int(math.ceil(max // 8)) + 1 # use at least 1 byte
785 if bytes_count > 32: # Don't encode more than 256 bits / values
786 bytes_count = 32
788 bitmap += struct.pack("BB", wb, bytes_count)
790 # Generate the bitmap
791 # The idea is to remove out of range Resource Records with these steps
792 # 1. rescale to fit into 8 bits
793 # 2. x gives the bit position ; compute the corresponding value
794 # 3. sum everything
795 bitmap += b"".join(
796 struct.pack(
797 b"B",
798 sum(2 ** (7 - (x - 256 * wb) + (tmp * 8)) for x in rrlist
799 if 256 * wb + 8 * tmp <= x < 256 * wb + 8 * tmp + 8),
800 ) for tmp in range(bytes_count)
801 )
803 return bitmap
806class RRlistField(StrField):
807 islist = 1
809 def h2i(self, pkt, x):
810 if x and isinstance(x, list):
811 return RRlist2bitmap(x)
812 return x
814 def i2repr(self, pkt, x):
815 if not x:
816 return "[]"
817 x = self.i2h(pkt, x)
818 rrlist = bitmap2RRlist(x)
819 return [dnstypes.get(rr, rr) for rr in rrlist] if rrlist else repr(x)
822class _DNSRRdummy(Packet):
823 name = "Dummy class that implements post_build() for Resource Records"
825 def post_build(self, pkt, pay):
826 if self.rdlen is not None:
827 return pkt + pay
829 lrrname = len(self.fields_desc[0].i2m("", self.getfieldval("rrname")))
830 tmp_len = len(pkt) - lrrname - 10
831 tmp_pkt = pkt[:lrrname + 8]
832 pkt = struct.pack("!H", tmp_len) + pkt[lrrname + 8 + 2:]
834 return tmp_pkt + pkt + pay
836 def default_payload_class(self, payload):
837 return conf.padding_layer
840class DNSRRHINFO(_DNSRRdummy):
841 name = "DNS HINFO Resource Record"
842 fields_desc = [DNSStrField("rrname", ""),
843 ShortEnumField("type", 13, dnstypes),
844 BitField("cacheflush", 0, 1), # mDNS RFC 6762
845 BitEnumField("rclass", 1, 15, dnsclasses),
846 IntField("ttl", 0),
847 ShortField("rdlen", None),
848 FieldLenField("cpulen", None, fmt="!B", length_of="cpu"),
849 StrLenField("cpu", "", length_from=lambda x: x.cpulen),
850 FieldLenField("oslen", None, fmt="!B", length_of="os"),
851 StrLenField("os", "", length_from=lambda x: x.oslen)]
854class DNSRRMX(_DNSRRdummy):
855 name = "DNS MX Resource Record"
856 fields_desc = [DNSStrField("rrname", ""),
857 ShortEnumField("type", 15, dnstypes),
858 BitField("cacheflush", 0, 1), # mDNS RFC 6762
859 BitEnumField("rclass", 1, 15, dnsclasses),
860 IntField("ttl", 0),
861 ShortField("rdlen", None),
862 ShortField("preference", 0),
863 DNSStrField("exchange", ""),
864 ]
867class DNSRRSOA(_DNSRRdummy):
868 name = "DNS SOA Resource Record"
869 fields_desc = [DNSStrField("rrname", ""),
870 ShortEnumField("type", 6, dnstypes),
871 ShortEnumField("rclass", 1, dnsclasses),
872 IntField("ttl", 0),
873 ShortField("rdlen", None),
874 DNSStrField("mname", ""),
875 DNSStrField("rname", ""),
876 IntField("serial", 0),
877 IntField("refresh", 0),
878 IntField("retry", 0),
879 IntField("expire", 0),
880 IntField("minimum", 0)
881 ]
884class DNSRRRSIG(_DNSRRdummy):
885 name = "DNS RRSIG Resource Record"
886 fields_desc = [DNSStrField("rrname", ""),
887 ShortEnumField("type", 46, dnstypes),
888 BitField("cacheflush", 0, 1), # mDNS RFC 6762
889 BitEnumField("rclass", 1, 15, dnsclasses),
890 IntField("ttl", 0),
891 ShortField("rdlen", None),
892 ShortEnumField("typecovered", 1, dnstypes),
893 ByteEnumField("algorithm", 5, dnssecalgotypes),
894 ByteField("labels", 0),
895 IntField("originalttl", 0),
896 UTCTimeField("expiration", 0),
897 UTCTimeField("inception", 0),
898 ShortField("keytag", 0),
899 DNSStrField("signersname", ""),
900 StrField("signature", "")
901 ]
904class DNSRRNSEC(_DNSRRdummy):
905 name = "DNS NSEC Resource Record"
906 fields_desc = [DNSStrField("rrname", ""),
907 ShortEnumField("type", 47, dnstypes),
908 BitField("cacheflush", 0, 1), # mDNS RFC 6762
909 BitEnumField("rclass", 1, 15, dnsclasses),
910 IntField("ttl", 0),
911 ShortField("rdlen", None),
912 DNSStrField("nextname", ""),
913 RRlistField("typebitmaps", [])
914 ]
917class DNSRRDNSKEY(_DNSRRdummy):
918 name = "DNS DNSKEY Resource Record"
919 fields_desc = [DNSStrField("rrname", ""),
920 ShortEnumField("type", 48, dnstypes),
921 BitField("cacheflush", 0, 1), # mDNS RFC 6762
922 BitEnumField("rclass", 1, 15, dnsclasses),
923 IntField("ttl", 0),
924 ShortField("rdlen", None),
925 FlagsField("flags", 256, 16, "S???????Z???????"),
926 # S: Secure Entry Point
927 # Z: Zone Key
928 ByteField("protocol", 3),
929 ByteEnumField("algorithm", 5, dnssecalgotypes),
930 StrField("publickey", "")
931 ]
934class DNSRRDS(_DNSRRdummy):
935 name = "DNS DS Resource Record"
936 fields_desc = [DNSStrField("rrname", ""),
937 ShortEnumField("type", 43, dnstypes),
938 BitField("cacheflush", 0, 1), # mDNS RFC 6762
939 BitEnumField("rclass", 1, 15, dnsclasses),
940 IntField("ttl", 0),
941 ShortField("rdlen", None),
942 ShortField("keytag", 0),
943 ByteEnumField("algorithm", 5, dnssecalgotypes),
944 ByteEnumField("digesttype", 5, dnssecdigesttypes),
945 StrField("digest", "")
946 ]
949# RFC 5074 - DNSSEC Lookaside Validation (DLV)
950class DNSRRDLV(DNSRRDS):
951 name = "DNS DLV Resource Record"
953 def __init__(self, *args, **kargs):
954 DNSRRDS.__init__(self, *args, **kargs)
955 if not kargs.get('type', 0):
956 self.type = 32769
958# RFC 5155 - DNS Security (DNSSEC) Hashed Authenticated Denial of Existence
961class DNSRRNSEC3(_DNSRRdummy):
962 name = "DNS NSEC3 Resource Record"
963 fields_desc = [DNSStrField("rrname", ""),
964 ShortEnumField("type", 50, dnstypes),
965 BitField("cacheflush", 0, 1), # mDNS RFC 6762
966 BitEnumField("rclass", 1, 15, dnsclasses),
967 IntField("ttl", 0),
968 ShortField("rdlen", None),
969 ByteField("hashalg", 0),
970 BitEnumField("flags", 0, 8, {1: "Opt-Out"}),
971 ShortField("iterations", 0),
972 FieldLenField("saltlength", 0, fmt="!B", length_of="salt"),
973 StrLenField("salt", "", length_from=lambda x: x.saltlength),
974 FieldLenField("hashlength", 0, fmt="!B", length_of="nexthashedownername"), # noqa: E501
975 StrLenField("nexthashedownername", "", length_from=lambda x: x.hashlength), # noqa: E501
976 RRlistField("typebitmaps", [])
977 ]
980class DNSRRNSEC3PARAM(_DNSRRdummy):
981 name = "DNS NSEC3PARAM Resource Record"
982 fields_desc = [DNSStrField("rrname", ""),
983 ShortEnumField("type", 51, dnstypes),
984 BitField("cacheflush", 0, 1), # mDNS RFC 6762
985 BitEnumField("rclass", 1, 15, dnsclasses),
986 IntField("ttl", 0),
987 ShortField("rdlen", None),
988 ByteField("hashalg", 0),
989 ByteField("flags", 0),
990 ShortField("iterations", 0),
991 FieldLenField("saltlength", 0, fmt="!B", length_of="salt"),
992 StrLenField("salt", "", length_from=lambda pkt: pkt.saltlength) # noqa: E501
993 ]
996# RFC 9460 Service Binding and Parameter Specification via the DNS
997# https://www.rfc-editor.org/rfc/rfc9460.html
1000# https://www.iana.org/assignments/dns-svcb/dns-svcb.xhtml
1001svc_param_keys = {
1002 0: "mandatory",
1003 1: "alpn",
1004 2: "no-default-alpn",
1005 3: "port",
1006 4: "ipv4hint",
1007 5: "ech",
1008 6: "ipv6hint",
1009 7: "dohpath",
1010 8: "ohttp",
1011}
1014class SvcParam(Packet):
1015 name = "SvcParam"
1016 fields_desc = [ShortEnumField("key", 0, svc_param_keys),
1017 FieldLenField("len", None, length_of="value", fmt="H"),
1018 MultipleTypeField(
1019 [
1020 # mandatory
1021 (FieldListField("value", [],
1022 ShortEnumField("", 0, svc_param_keys),
1023 length_from=lambda pkt: pkt.len),
1024 lambda pkt: pkt.key == 0),
1025 # alpn, no-default-alpn
1026 (DNSTextField("value", [],
1027 length_from=lambda pkt: pkt.len),
1028 lambda pkt: pkt.key in (1, 2)),
1029 # port
1030 (ShortField("value", 0),
1031 lambda pkt: pkt.key == 3),
1032 # ipv4hint
1033 (FieldListField("value", [],
1034 IPField("", "0.0.0.0"),
1035 length_from=lambda pkt: pkt.len),
1036 lambda pkt: pkt.key == 4),
1037 # ipv6hint
1038 (FieldListField("value", [],
1039 IP6Field("", "::"),
1040 length_from=lambda pkt: pkt.len),
1041 lambda pkt: pkt.key == 6),
1042 ],
1043 StrLenField("value", "",
1044 length_from=lambda pkt:pkt.len))]
1046 def extract_padding(self, p):
1047 return "", p
1050class DNSRRSVCB(_DNSRRdummy):
1051 name = "DNS SVCB Resource Record"
1052 fields_desc = [DNSStrField("rrname", ""),
1053 ShortEnumField("type", 64, dnstypes),
1054 BitField("cacheflush", 0, 1), # mDNS RFC 6762
1055 BitEnumField("rclass", 1, 15, dnsclasses),
1056 IntField("ttl", 0),
1057 ShortField("rdlen", None),
1058 ShortField("svc_priority", 0),
1059 DNSStrField("target_name", ""),
1060 PacketListField("svc_params", [], SvcParam)]
1063class DNSRRHTTPS(_DNSRRdummy):
1064 name = "DNS HTTPS Resource Record"
1065 fields_desc = [DNSStrField("rrname", ""),
1066 ShortEnumField("type", 65, dnstypes)
1067 ] + DNSRRSVCB.fields_desc[2:]
1070# RFC 2782 - A DNS RR for specifying the location of services (DNS SRV)
1073class DNSRRSRV(_DNSRRdummy):
1074 name = "DNS SRV Resource Record"
1075 fields_desc = [DNSStrField("rrname", ""),
1076 ShortEnumField("type", 33, dnstypes),
1077 BitField("cacheflush", 0, 1), # mDNS RFC 6762
1078 BitEnumField("rclass", 1, 15, dnsclasses),
1079 IntField("ttl", 0),
1080 ShortField("rdlen", None),
1081 ShortField("priority", 0),
1082 ShortField("weight", 0),
1083 ShortField("port", 0),
1084 DNSStrField("target", ""), ]
1087# RFC 2845 - Secret Key Transaction Authentication for DNS (TSIG)
1088tsig_algo_sizes = {"HMAC-MD5.SIG-ALG.REG.INT": 16,
1089 "hmac-sha1": 20}
1092class TimeSignedField(Field[int, bytes]):
1093 def __init__(self, name, default):
1094 Field.__init__(self, name, default, fmt="6s")
1096 def _convert_seconds(self, packed_seconds):
1097 """Unpack the internal representation."""
1098 seconds = struct.unpack("!H", packed_seconds[:2])[0]
1099 seconds += struct.unpack("!I", packed_seconds[2:])[0]
1100 return seconds
1102 def i2m(self, pkt, seconds):
1103 """Convert the number of seconds since 1-Jan-70 UTC to the packed
1104 representation."""
1106 if seconds is None:
1107 seconds = 0
1109 tmp_short = (seconds >> 32) & 0xFFFF
1110 tmp_int = seconds & 0xFFFFFFFF
1112 return struct.pack("!HI", tmp_short, tmp_int)
1114 def m2i(self, pkt, packed_seconds):
1115 """Convert the internal representation to the number of seconds
1116 since 1-Jan-70 UTC."""
1118 if packed_seconds is None:
1119 return None
1121 return self._convert_seconds(packed_seconds)
1123 def i2repr(self, pkt, packed_seconds):
1124 """Convert the internal representation to a nice one using the RFC
1125 format."""
1126 time_struct = time.gmtime(packed_seconds)
1127 return time.strftime("%a %b %d %H:%M:%S %Y", time_struct)
1130class DNSRRTSIG(_DNSRRdummy):
1131 name = "DNS TSIG Resource Record"
1132 fields_desc = [DNSStrField("rrname", ""),
1133 ShortEnumField("type", 250, dnstypes),
1134 ShortEnumField("rclass", 1, dnsclasses),
1135 IntField("ttl", 0),
1136 ShortField("rdlen", None),
1137 DNSStrField("algo_name", "hmac-sha1"),
1138 TimeSignedField("time_signed", 0),
1139 ShortField("fudge", 0),
1140 FieldLenField("mac_len", None, fmt="!H", length_of="mac_data"), # noqa: E501
1141 StrLenField("mac_data", "", length_from=lambda pkt: pkt.mac_len), # noqa: E501
1142 ShortField("original_id", 0),
1143 ShortField("error", 0),
1144 FieldLenField("other_len", None, fmt="!H", length_of="other_data"), # noqa: E501
1145 StrLenField("other_data", "", length_from=lambda pkt: pkt.other_len) # noqa: E501
1146 ]
1149class DNSRRNAPTR(_DNSRRdummy):
1150 name = "DNS NAPTR Resource Record"
1151 fields_desc = [DNSStrField("rrname", ""),
1152 ShortEnumField("type", 35, dnstypes),
1153 BitField("cacheflush", 0, 1), # mDNS RFC 6762
1154 BitEnumField("rclass", 1, 15, dnsclasses),
1155 IntField("ttl", 0),
1156 ShortField("rdlen", None),
1157 ShortField("order", 0),
1158 ShortField("preference", 0),
1159 FieldLenField("flags_len", None, fmt="!B", length_of="flags"),
1160 StrLenField("flags", "", length_from=lambda pkt: pkt.flags_len),
1161 FieldLenField("services_len", None, fmt="!B", length_of="services"),
1162 StrLenField("services", "",
1163 length_from=lambda pkt: pkt.services_len),
1164 FieldLenField("regexp_len", None, fmt="!B", length_of="regexp"),
1165 StrLenField("regexp", "", length_from=lambda pkt: pkt.regexp_len),
1166 DNSStrField("replacement", ""),
1167 ]
1170DNSRR_DISPATCHER = {
1171 6: DNSRRSOA, # RFC 1035
1172 13: DNSRRHINFO, # RFC 1035
1173 15: DNSRRMX, # RFC 1035
1174 33: DNSRRSRV, # RFC 2782
1175 35: DNSRRNAPTR, # RFC 2915
1176 41: DNSRROPT, # RFC 1671
1177 43: DNSRRDS, # RFC 4034
1178 46: DNSRRRSIG, # RFC 4034
1179 47: DNSRRNSEC, # RFC 4034
1180 48: DNSRRDNSKEY, # RFC 4034
1181 50: DNSRRNSEC3, # RFC 5155
1182 51: DNSRRNSEC3PARAM, # RFC 5155
1183 64: DNSRRSVCB, # RFC 9460
1184 65: DNSRRHTTPS, # RFC 9460
1185 250: DNSRRTSIG, # RFC 2845
1186 32769: DNSRRDLV, # RFC 4431
1187}
1190class DNSRR(Packet):
1191 name = "DNS Resource Record"
1192 show_indent = 0
1193 fields_desc = [DNSStrField("rrname", ""),
1194 ShortEnumField("type", 1, dnstypes),
1195 BitField("cacheflush", 0, 1), # mDNS RFC 6762
1196 BitEnumField("rclass", 1, 15, dnsclasses),
1197 IntField("ttl", 0),
1198 FieldLenField("rdlen", None, length_of="rdata", fmt="H"),
1199 MultipleTypeField(
1200 [
1201 # A
1202 (IPField("rdata", "0.0.0.0"),
1203 lambda pkt: pkt.type == 1),
1204 # AAAA
1205 (IP6Field("rdata", "::"),
1206 lambda pkt: pkt.type == 28),
1207 # NS, MD, MF, CNAME, PTR, DNAME
1208 (DNSStrField("rdata", "",
1209 length_from=lambda pkt: pkt.rdlen),
1210 lambda pkt: pkt.type in [2, 3, 4, 5, 12, 39]),
1211 # TEXT
1212 (DNSTextField("rdata", [""],
1213 length_from=lambda pkt: pkt.rdlen),
1214 lambda pkt: pkt.type == 16),
1215 ],
1216 StrLenField("rdata", "",
1217 length_from=lambda pkt:pkt.rdlen)
1218 )]
1220 def default_payload_class(self, payload):
1221 return conf.padding_layer
1224def _DNSRR(s, **kwargs):
1225 """
1226 DNSRR dispatcher func
1227 """
1228 if s:
1229 # Try to find the type of the RR using the dispatcher
1230 _, remain = dns_get_str(s, _ignore_compression=True)
1231 cls = DNSRR_DISPATCHER.get(
1232 struct.unpack("!H", remain[:2])[0],
1233 DNSRR,
1234 )
1235 rrlen = (
1236 len(s) - len(remain) + # rrname len
1237 10 +
1238 struct.unpack("!H", remain[8:10])[0]
1239 )
1240 pkt = cls(s[:rrlen], **kwargs) / conf.padding_layer(s[rrlen:])
1241 # drop rdlen because if rdata was compressed, it will break everything
1242 # when rebuilding
1243 del pkt.fields["rdlen"]
1244 return pkt
1245 return None
1248class DNSQR(Packet):
1249 name = "DNS Question Record"
1250 show_indent = 0
1251 fields_desc = [DNSStrField("qname", "www.example.com"),
1252 ShortEnumField("qtype", 1, dnsqtypes),
1253 BitField("unicastresponse", 0, 1), # mDNS RFC 6762
1254 BitEnumField("qclass", 1, 15, dnsclasses)]
1256 def default_payload_class(self, payload):
1257 return conf.padding_layer
1260class _DNSPacketListField(PacketListField):
1261 # A normal PacketListField with backward-compatible hacks
1262 def any2i(self, pkt, x):
1263 # type: (Optional[Packet], List[Any]) -> List[Any]
1264 if x is None:
1265 warnings.warn(
1266 ("The DNS fields 'qd', 'an', 'ns' and 'ar' are now "
1267 "PacketListField(s) ! "
1268 "Setting a null default should be [] instead of None"),
1269 DeprecationWarning
1270 )
1271 x = []
1272 return super(_DNSPacketListField, self).any2i(pkt, x)
1274 def i2h(self, pkt, x):
1275 # type: (Optional[Packet], List[Packet]) -> Any
1276 class _list(list):
1277 """
1278 Fake list object to provide compatibility with older DNS fields
1279 """
1280 def __getattr__(self, attr):
1281 try:
1282 ret = getattr(self[0], attr)
1283 warnings.warn(
1284 ("The DNS fields 'qd', 'an', 'ns' and 'ar' are now "
1285 "PacketListField(s) ! "
1286 "To access the first element, use pkt.an[0] instead of "
1287 "pkt.an"),
1288 DeprecationWarning
1289 )
1290 return ret
1291 except AttributeError:
1292 raise
1293 return _list(x)
1296class DNS(DNSCompressedPacket):
1297 name = "DNS"
1298 FORCE_TCP = False
1299 fields_desc = [
1300 ConditionalField(ShortField("length", None),
1301 lambda p: p.FORCE_TCP or isinstance(p.underlayer, TCP)),
1302 ShortField("id", 0),
1303 BitField("qr", 0, 1),
1304 BitEnumField("opcode", 0, 4, {0: "QUERY", 1: "IQUERY", 2: "STATUS"}),
1305 BitField("aa", 0, 1),
1306 BitField("tc", 0, 1),
1307 BitField("rd", 1, 1),
1308 BitField("ra", 0, 1),
1309 BitField("z", 0, 1),
1310 # AD and CD bits are defined in RFC 2535
1311 BitField("ad", 0, 1), # Authentic Data
1312 BitField("cd", 0, 1), # Checking Disabled
1313 BitEnumField("rcode", 0, 4, {0: "ok", 1: "format-error",
1314 2: "server-failure", 3: "name-error",
1315 4: "not-implemented", 5: "refused"}),
1316 FieldLenField("qdcount", None, count_of="qd"),
1317 FieldLenField("ancount", None, count_of="an"),
1318 FieldLenField("nscount", None, count_of="ns"),
1319 FieldLenField("arcount", None, count_of="ar"),
1320 _DNSPacketListField("qd", [DNSQR()], DNSQR, count_from=lambda pkt: pkt.qdcount),
1321 _DNSPacketListField("an", [], _DNSRR, count_from=lambda pkt: pkt.ancount),
1322 _DNSPacketListField("ns", [], _DNSRR, count_from=lambda pkt: pkt.nscount),
1323 _DNSPacketListField("ar", [], _DNSRR, count_from=lambda pkt: pkt.arcount),
1324 ]
1326 def get_full(self):
1327 # Required for DNSCompressedPacket
1328 if isinstance(self.underlayer, TCP) or self.FORCE_TCP:
1329 return self.original[2:]
1330 else:
1331 return self.original
1333 def answers(self, other):
1334 return (isinstance(other, DNS) and
1335 self.id == other.id and
1336 self.qr == 1 and
1337 other.qr == 0)
1339 def mysummary(self):
1340 name = ""
1341 if self.qr:
1342 type = "Ans"
1343 if self.an and isinstance(self.an[0], DNSRR):
1344 name = ' %s' % self.an[0].rdata
1345 elif self.rcode != 0:
1346 name = self.sprintf(' %rcode%')
1347 else:
1348 type = "Qry"
1349 if self.qd and isinstance(self.qd[0], DNSQR):
1350 name = ' %s' % self.qd[0].qname
1351 return "%sDNS %s%s" % (
1352 "m"
1353 if isinstance(self.underlayer, UDP) and self.underlayer.dport == 5353
1354 else "",
1355 type,
1356 name,
1357 )
1359 def post_build(self, pkt, pay):
1360 if (
1361 (isinstance(self.underlayer, TCP) or self.FORCE_TCP) and
1362 self.length is None
1363 ):
1364 pkt = struct.pack("!H", len(pkt) - 2) + pkt[2:]
1365 return pkt + pay
1367 def compress(self):
1368 """Return the compressed DNS packet (using `dns_compress()`)"""
1369 return dns_compress(self)
1371 def pre_dissect(self, s):
1372 """
1373 Check that a valid DNS over TCP message can be decoded
1374 """
1375 if isinstance(self.underlayer, TCP):
1377 # Compute the length of the DNS packet
1378 if len(s) >= 2:
1379 dns_len = struct.unpack("!H", s[:2])[0]
1380 else:
1381 message = "Malformed DNS message: too small!"
1382 log_runtime.info(message)
1383 raise Scapy_Exception(message)
1385 # Check if the length is valid
1386 if dns_len < 14 or len(s) < dns_len:
1387 message = "Malformed DNS message: invalid length!"
1388 log_runtime.info(message)
1389 raise Scapy_Exception(message)
1391 return s
1394class DNSTCP(DNS):
1395 """
1396 A DNS packet that is always under TCP
1397 """
1398 FORCE_TCP = True
1399 match_subclass = True
1402bind_layers(UDP, DNS, dport=5353)
1403bind_layers(UDP, DNS, sport=5353)
1404bind_layers(UDP, DNS, dport=53)
1405bind_layers(UDP, DNS, sport=53)
1406DestIPField.bind_addr(UDP, "224.0.0.251", dport=5353)
1407if conf.ipv6_enabled:
1408 from scapy.layers.inet6 import DestIP6Field
1409 DestIP6Field.bind_addr(UDP, "ff02::fb", dport=5353)
1410bind_layers(TCP, DNS, dport=53)
1411bind_layers(TCP, DNS, sport=53)
1413# Nameserver config
1414conf.nameservers = read_nameservers()
1415_dns_cache = conf.netcache.new_cache("dns_cache", 300)
1418@conf.commands.register
1419def dns_resolve(qname, qtype="A", raw=False, tcp=False, verbose=1, timeout=3, **kwargs):
1420 """
1421 Perform a simple DNS resolution using conf.nameservers with caching
1423 :param qname: the name to query
1424 :param qtype: the type to query (default A)
1425 :param raw: return the whole DNS packet (default False)
1426 :param tcp: whether to use directly TCP instead of UDP. If truncated is received,
1427 UDP automatically retries in TCP. (default: False)
1428 :param verbose: show verbose errors
1429 :param timeout: seconds until timeout (per server)
1430 :raise TimeoutError: if no DNS servers were reached in time.
1431 """
1432 # Unify types (for caching)
1433 qtype = DNSQR.qtype.any2i_one(None, qtype)
1434 qname = DNSQR.qname.any2i(None, qname)
1435 # Check cache
1436 cache_ident = b";".join(
1437 [qname, struct.pack("!H", qtype)] +
1438 ([b"raw"] if raw else [])
1439 )
1440 result = _dns_cache.get(cache_ident)
1441 if result:
1442 return result
1444 kwargs.setdefault("timeout", timeout)
1445 kwargs.setdefault("verbose", 0) # hide sr1() output
1446 res = None
1447 for nameserver in conf.nameservers:
1448 # Try all nameservers
1449 try:
1450 # Spawn a socket, connect to the nameserver on port 53
1451 if tcp:
1452 cls = DNSTCP
1453 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1454 else:
1455 cls = DNS
1456 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1457 sock.settimeout(kwargs["timeout"])
1458 sock.connect((nameserver, 53))
1459 # Connected. Wrap it with DNS
1460 sock = StreamSocket(sock, cls)
1461 # I/O
1462 res = sock.sr1(
1463 cls(qd=[DNSQR(qname=qname, qtype=qtype)], id=RandShort()),
1464 **kwargs,
1465 )
1466 except IOError as ex:
1467 if verbose:
1468 log_runtime.warning(str(ex))
1469 continue
1470 finally:
1471 sock.close()
1472 if res:
1473 # We have a response ! Check for failure
1474 if res[DNS].tc == 1: # truncated !
1475 if not tcp:
1476 # Retry using TCP
1477 return dns_resolve(
1478 qname=qname,
1479 qtype=qtype,
1480 raw=raw,
1481 tcp=True,
1482 **kwargs,
1483 )
1484 elif verbose:
1485 log_runtime.info("DNS answer is truncated !")
1487 if res[DNS].rcode == 2: # server failure
1488 res = None
1489 if verbose:
1490 log_runtime.info(
1491 "DNS: %s answered with failure for %s" % (
1492 nameserver,
1493 qname,
1494 )
1495 )
1496 else:
1497 break
1498 if res is not None:
1499 if raw:
1500 # Raw
1501 result = res
1502 else:
1503 # Find answers
1504 result = [
1505 x
1506 for x in itertools.chain(res.an, res.ns, res.ar)
1507 if getattr(x, "type", None) == qtype
1508 ]
1509 if result:
1510 # Cache it
1511 _dns_cache[cache_ident] = result
1512 return result
1513 else:
1514 raise TimeoutError
1517@conf.commands.register
1518def dyndns_add(nameserver, name, rdata, type="A", ttl=10):
1519 """Send a DNS add message to a nameserver for "name" to have a new "rdata"
1520dyndns_add(nameserver, name, rdata, type="A", ttl=10) -> result code (0=ok)
1522example: dyndns_add("ns1.toto.com", "dyn.toto.com", "127.0.0.1")
1523RFC2136
1524"""
1525 zone = name[name.find(".") + 1:]
1526 r = sr1(IP(dst=nameserver) / UDP() / DNS(opcode=5,
1527 qd=[DNSQR(qname=zone, qtype="SOA")], # noqa: E501
1528 ns=[DNSRR(rrname=name, type="A",
1529 ttl=ttl, rdata=rdata)]),
1530 verbose=0, timeout=5)
1531 if r and r.haslayer(DNS):
1532 return r.getlayer(DNS).rcode
1533 else:
1534 return -1
1537@conf.commands.register
1538def dyndns_del(nameserver, name, type="ALL", ttl=10):
1539 """Send a DNS delete message to a nameserver for "name"
1540dyndns_del(nameserver, name, type="ANY", ttl=10) -> result code (0=ok)
1542example: dyndns_del("ns1.toto.com", "dyn.toto.com")
1543RFC2136
1544"""
1545 zone = name[name.find(".") + 1:]
1546 r = sr1(IP(dst=nameserver) / UDP() / DNS(opcode=5,
1547 qd=[DNSQR(qname=zone, qtype="SOA")], # noqa: E501
1548 ns=[DNSRR(rrname=name, type=type,
1549 rclass="ANY", ttl=0, rdata="")]), # noqa: E501
1550 verbose=0, timeout=5)
1551 if r and r.haslayer(DNS):
1552 return r.getlayer(DNS).rcode
1553 else:
1554 return -1
1557class DNS_am(AnsweringMachine):
1558 function_name = "dnsd"
1559 filter = "udp port 53"
1560 cls = DNS # We also use this automaton for llmnrd / mdnsd
1562 def parse_options(self, joker=None,
1563 match=None,
1564 srvmatch=None,
1565 joker6=False,
1566 send_error=False,
1567 relay=False,
1568 from_ip=True,
1569 from_ip6=False,
1570 src_ip=None,
1571 src_ip6=None,
1572 ttl=10,
1573 jokerarpa=False):
1574 """
1575 Simple DNS answering machine.
1577 :param joker: default IPv4 for unresolved domains.
1578 Set to False to disable, None to mirror the interface's IP.
1579 Defaults to None, unless 'match' is used, then it defaults to
1580 False.
1581 :param joker6: default IPv6 for unresolved domains.
1582 Set to False to disable, None to mirror the interface's IPv6.
1583 Defaults to False.
1584 :param match: queries to match.
1585 This can be a dictionary of {name: val} where name is a string
1586 representing a domain name (A, AAAA) and val is a tuple of 2
1587 elements, each representing an IP or a list of IPs. If val is
1588 a single element, (A, None) is assumed.
1589 This can also be a list or names, in which case joker(6) are
1590 used as a response.
1591 :param jokerarpa: answer for .in-addr.arpa PTR requests. (Default: False)
1592 :param relay: relay unresolved domains to conf.nameservers (Default: False).
1593 :param send_error: send an error message when this server can't answer
1594 (Default: False)
1595 :param srvmatch: a dictionary of {name: (port, target)} used for SRV
1596 :param from_ip: an source IP to filter. Can contain a netmask. True for all,
1597 False for none. Default True
1598 :param from_ip6: an source IPv6 to filter. Can contain a netmask. True for all,
1599 False for none. Default False
1600 :param ttl: the DNS time to live (in seconds)
1601 :param src_ip: override the source IP
1602 :param src_ip6:
1604 Examples:
1606 - Answer all 'A' and 'AAAA' requests::
1608 $ sudo iptables -I OUTPUT -p icmp --icmp-type 3/3 -j DROP
1609 >>> dnsd(joker="192.168.0.2", joker6="fe80::260:8ff:fe52:f9d8",
1610 ... iface="eth0")
1612 - Answer only 'A' query for google.com with 192.168.0.2::
1614 >>> dnsd(match={"google.com": "192.168.0.2"}, iface="eth0")
1616 - Answer DNS for a Windows domain controller ('SRV', 'A' and 'AAAA')::
1618 >>> dnsd(
1619 ... srvmatch={
1620 ... "_ldap._tcp.dc._msdcs.DOMAIN.LOCAL.": (389,
1621 ... "srv1.domain.local"),
1622 ... },
1623 ... match={"src1.domain.local": ("192.168.0.102",
1624 ... "fe80::260:8ff:fe52:f9d8")},
1625 ... )
1627 - Relay all queries to another DNS server, except some::
1629 >>> conf.nameservers = ["1.1.1.1"] # server to relay to
1630 >>> dnsd(
1631 ... match={"test.com": "1.1.1.1"},
1632 ... relay=True,
1633 ... )
1634 """
1635 from scapy.layers.inet6 import Net6
1637 self.mDNS = isinstance(self, mDNS_am)
1638 self.llmnr = self.cls != DNS
1640 # Add some checks (to help)
1641 if not isinstance(joker, (str, bool)) and joker is not None:
1642 raise ValueError("Bad 'joker': should be an IPv4 (str) or False !")
1643 if not isinstance(joker6, (str, bool)) and joker6 is not None:
1644 raise ValueError("Bad 'joker6': should be an IPv6 (str) or False !")
1645 if not isinstance(jokerarpa, (str, bool)):
1646 raise ValueError("Bad 'jokerarpa': should be a hostname or False !")
1647 if not isinstance(from_ip, (str, Net, bool)):
1648 raise ValueError("Bad 'from_ip': should be an IPv4 (str), Net or False !")
1649 if not isinstance(from_ip6, (str, Net6, bool)):
1650 raise ValueError("Bad 'from_ip6': should be an IPv6 (str), Net or False !")
1651 if self.mDNS and src_ip:
1652 raise ValueError("Cannot use 'src_ip' in mDNS !")
1653 if self.mDNS and src_ip6:
1654 raise ValueError("Cannot use 'src_ip6' in mDNS !")
1656 if joker is None and match is not None:
1657 joker = False
1658 self.joker = joker
1659 self.joker6 = joker6
1660 self.jokerarpa = jokerarpa
1662 def normv(v):
1663 if isinstance(v, (tuple, list)) and len(v) == 2:
1664 return tuple(v)
1665 elif isinstance(v, str):
1666 return (v, joker6)
1667 else:
1668 raise ValueError("Bad match value: '%s'" % repr(v))
1670 def normk(k):
1671 k = bytes_encode(k).lower()
1672 if not k.endswith(b"."):
1673 k += b"."
1674 return k
1676 self.match = {}
1677 if match:
1678 if isinstance(match, (list, set)):
1679 self.match.update({normk(k): (None, None) for k in match})
1680 else:
1681 self.match.update({normk(k): normv(v) for k, v in match.items()})
1682 if srvmatch is None:
1683 self.srvmatch = {}
1684 else:
1685 self.srvmatch = {normk(k): normv(v) for k, v in srvmatch.items()}
1687 self.send_error = send_error
1688 self.relay = relay
1689 if isinstance(from_ip, str):
1690 self.from_ip = Net(from_ip)
1691 else:
1692 self.from_ip = from_ip
1693 if isinstance(from_ip6, str):
1694 self.from_ip6 = Net6(from_ip6)
1695 else:
1696 self.from_ip6 = from_ip6
1697 self.src_ip = src_ip
1698 self.src_ip6 = src_ip6
1699 self.ttl = ttl
1701 def is_request(self, req):
1702 from scapy.layers.inet6 import IPv6
1703 query = req.getlayer(self.cls)
1704 if query is None or query.qr != 0:
1705 return False
1706 if self.llmnr:
1707 # [RFC 4795] sect 2.1: an LLMNR query goes to the link-scope
1708 # multicast address with a hop limit of 1, so a responder must
1709 # not answer one that arrived any other way. mDNS is left alone:
1710 # it permits a direct unicast query and LLMNR does not.
1711 network = query.underlayer.underlayer
1712 if isinstance(network, IPv6):
1713 if network.dst != "ff02::1:3" or network.hlim != 1:
1714 return False
1715 elif isinstance(network, IP):
1716 if network.dst != "224.0.0.252" or network.ttl != 1:
1717 return False
1718 else:
1719 return False
1720 return (
1721 (
1722 self.from_ip6 is True or
1723 (self.from_ip6 and req[IPv6].src in self.from_ip6)
1724 )
1725 if IPv6 in req else
1726 (
1727 self.from_ip is True or
1728 (self.from_ip and req[IP].src in self.from_ip)
1729 )
1730 )
1732 def make_reply(self, req):
1733 # Build reply from the request
1734 resp = req.copy()
1735 if Ether in req:
1736 if self.mDNS:
1737 resp[Ether].src, resp[Ether].dst = None, None
1738 elif self.llmnr:
1739 resp[Ether].src, resp[Ether].dst = None, req[Ether].src
1740 else:
1741 resp[Ether].src, resp[Ether].dst = (
1742 None if req[Ether].dst == "ff:ff:ff:ff:ff:ff" else req[Ether].dst,
1743 req[Ether].src,
1744 )
1745 from scapy.layers.inet6 import IPv6
1746 if IPv6 in req:
1747 resp[IPv6].underlayer.remove_payload()
1748 if self.mDNS:
1749 # "All Multicast DNS responses (including responses sent via unicast)
1750 # SHOULD be sent with IP TTL set to 255."
1751 resp /= IPv6(dst="ff02::fb", src=self.src_ip6,
1752 fl=req[IPv6].fl, hlim=255)
1753 elif self.llmnr:
1754 resp /= IPv6(dst=req[IPv6].src, src=self.src_ip6,
1755 fl=req[IPv6].fl, hlim=req[IPv6].hlim)
1756 else:
1757 resp /= IPv6(dst=req[IPv6].src, src=self.src_ip6 or req[IPv6].dst,
1758 fl=req[IPv6].fl, hlim=req[IPv6].hlim)
1759 elif IP in req:
1760 resp[IP].underlayer.remove_payload()
1761 if self.mDNS:
1762 # "All Multicast DNS responses (including responses sent via unicast)
1763 # SHOULD be sent with IP TTL set to 255."
1764 resp /= IP(dst="224.0.0.251", src=self.src_ip,
1765 id=req[IP].id, ttl=255)
1766 elif self.llmnr:
1767 resp /= IP(dst=req[IP].src, src=self.src_ip,
1768 id=req[IP].id, ttl=req[IP].ttl)
1769 else:
1770 resp /= IP(dst=req[IP].src, src=self.src_ip or req[IP].dst,
1771 id=req[IP].id, ttl=req[IP].ttl)
1772 else:
1773 warning("No IP or IPv6 layer in %s", req.command())
1774 return
1775 try:
1776 resp /= UDP(sport=req[UDP].dport, dport=req[UDP].sport)
1777 except IndexError:
1778 warning("No UDP layer in %s", req.command(), exc_info=True)
1779 return
1780 try:
1781 req = req[self.cls]
1782 except IndexError:
1783 warning(
1784 "No %s layer in %s",
1785 self.cls.__name__,
1786 req.command(),
1787 exc_info=True,
1788 )
1789 return
1790 try:
1791 queries = req.qd
1792 except AttributeError:
1793 warning("No qd attribute in %s", req.command(), exc_info=True)
1794 return
1795 # Special case: alias 'ALL' query as 'A' + 'AAAA'
1796 try:
1797 allquery = next(
1798 (x for x in queries if getattr(x, "qtype", None) == 255)
1799 )
1800 queries.remove(allquery)
1801 queries.extend([
1802 DNSQR(
1803 qtype=x,
1804 qname=allquery.qname,
1805 unicastresponse=allquery.unicastresponse,
1806 qclass=allquery.qclass,
1807 )
1808 for x in [1, 28]
1809 ])
1810 except StopIteration:
1811 pass
1812 # Process each query
1813 ans = []
1814 ars = []
1815 for rq in queries:
1816 if isinstance(rq, Raw):
1817 warning("Cannot parse qd element %s", rq.command(), exc_info=True)
1818 continue
1819 rqname = rq.qname.lower()
1820 if rq.qtype in [1, 28]:
1821 # A or AAAA
1822 if rq.qtype == 28:
1823 # AAAA
1824 try:
1825 rdata = self.match[rqname][1]
1826 except KeyError:
1827 rdata = self.joker6
1828 if rdata is None and not self.relay:
1829 # 'None' resolves to the default IPv6
1830 iface = resolve_iface(self.optsniff.get("iface", conf.iface))
1831 if self.mDNS:
1832 # All IPs, as per mDNS.
1833 rdata = iface.ips[6]
1834 else:
1835 rdata = get_if_addr6(
1836 iface
1837 )
1838 if self.mDNS and rdata and IPv6 in resp:
1839 # For mDNS, we must replace the IPv6 src
1840 resp[IPv6].src = rdata
1841 elif rq.qtype == 1:
1842 # A
1843 try:
1844 rdata = self.match[rqname][0]
1845 except KeyError:
1846 rdata = self.joker
1847 if rdata is None and not self.relay:
1848 # 'None' resolves to the default IPv4
1849 iface = resolve_iface(self.optsniff.get("iface", conf.iface))
1850 if self.mDNS:
1851 # All IPs, as per mDNS.
1852 rdata = iface.ips[4]
1853 else:
1854 rdata = get_if_addr(
1855 iface
1856 )
1857 if self.mDNS and rdata and IP in resp:
1858 # For mDNS, we must replace the IP src
1859 resp[IP].src = rdata
1860 if rdata:
1861 # Common A and AAAA
1862 if not isinstance(rdata, list):
1863 rdata = [rdata]
1864 ans.extend([
1865 DNSRR(
1866 rrname=rq.qname,
1867 ttl=self.ttl,
1868 rdata=x,
1869 type=rq.qtype,
1870 cacheflush=self.mDNS and rq.qtype == rq.qtype,
1871 )
1872 for x in rdata
1873 ])
1874 continue # next
1875 elif rq.qtype == 33:
1876 # SRV
1877 try:
1878 port, target = self.srvmatch[rqname]
1879 ans.append(DNSRRSRV(
1880 rrname=rq.qname,
1881 port=port,
1882 target=target,
1883 weight=100,
1884 ttl=self.ttl
1885 ))
1886 continue # next
1887 except KeyError:
1888 # No result
1889 pass
1890 elif rq.qtype == 12:
1891 # PTR
1892 if rq.qname[-14:] == b".in-addr.arpa." and self.jokerarpa:
1893 ans.append(DNSRR(
1894 rrname=rq.qname,
1895 type=rq.qtype,
1896 ttl=self.ttl,
1897 rdata=self.jokerarpa,
1898 ))
1899 continue
1900 # It it arrives here, there is currently no answer
1901 if self.relay:
1902 # Relay mode ?
1903 try:
1904 _rslv = dns_resolve(rq.qname, qtype=rq.qtype, raw=True)
1905 if _rslv:
1906 ans.extend(_rslv.an)
1907 ars.extend(_rslv.ar)
1908 continue # next
1909 except TimeoutError:
1910 pass
1911 # Still no answer.
1912 if self.mDNS:
1913 # "Any time a responder receives a query for a name for which it
1914 # has verified exclusive ownership, for a type for which that name
1915 # has no records, the responder MUST respond asserting the
1916 # nonexistence of that record using a DNS NSEC record [RFC4034]."
1917 ans.append(DNSRRNSEC(
1918 # RFC6762 sect 6.1 - Negative Response
1919 ttl=self.ttl,
1920 rrname=rq.qname,
1921 nextname=rq.qname,
1922 typebitmaps=RRlist2bitmap([rq.qtype]),
1923 ))
1924 if self.mDNS and all(x.type == 47 for x in ans):
1925 # If mDNS answers with only NSEC, discard.
1926 return
1927 if not ans:
1928 # No answer is available.
1929 if self.send_error:
1930 resp /= self.cls(id=req.id, qr=1, qd=req.qd, rcode=3)
1931 return resp
1932 log_runtime.info("No answer could be provided to: %s" % req.summary())
1933 return
1934 # Handle Additional Records
1935 if self.mDNS:
1936 # Windows specific extension
1937 ars.append(DNSRROPT(
1938 z=0x1194,
1939 rdata=[
1940 EDNS0OWN(
1941 primary_mac=resp[Ether].src,
1942 ),
1943 ],
1944 ))
1945 # All rq were answered
1946 if self.mDNS:
1947 # in mDNS mode, don't repeat the question, set aa=1, rd=0
1948 dns = self.cls(id=req.id, aa=1, rd=0, qr=1, qd=[], ar=ars, an=ans)
1949 else:
1950 dns = self.cls(id=req.id, qr=1, qd=req.qd, ar=ars, an=ans)
1951 # Compress DNS and mDNS
1952 if not self.llmnr:
1953 resp /= dns_compress(dns)
1954 else:
1955 resp /= dns
1956 return resp
1959class mDNS_am(DNS_am):
1960 """
1961 mDNS answering machine.
1963 This has the same arguments as DNS_am. See help(DNS_am)
1965 Example::
1967 - Answer for 'TEST.local' with local IPv4::
1969 >>> mdnsd(match=["TEST.local"])
1971 - Answer all requests with other IP::
1973 >>> mdnsd(joker="192.168.0.2", joker6="fe80::260:8ff:fe52:f9d8",
1974 ... iface="eth0")
1976 - Answer for multiple different mDNS names::
1978 >>> mdnsd(match={"TEST.local": "192.168.0.100",
1979 ... "BOB.local": "192.168.0.101"})
1981 - Answer with both A and AAAA records::
1983 >>> mdnsd(match={"TEST.local": ("192.168.0.100",
1984 ... "fe80::260:8ff:fe52:f9d8")})
1985 """
1986 function_name = "mdnsd"
1987 filter = "udp port 5353"
1990# DNS-SD (RFC 6763)
1993class DNSSDResult(SndRcvList):
1994 def __init__(self,
1995 res=None, # type: Optional[Union[_PacketList[QueryAnswer], List[QueryAnswer]]] # noqa: E501
1996 name="DNS-SD", # type: str
1997 stats=None # type: Optional[List[Type[Packet]]]
1998 ):
1999 SndRcvList.__init__(self, res, name, stats)
2001 def show(self, types=['PTR', 'SRV'], alltypes=False):
2002 # type: (List[str], bool) -> None
2003 """
2004 Print the list of discovered services.
2006 :param types: types to show. Default ['PTR', 'SRV']
2007 :param alltypes: show all types. Default False
2008 """
2009 if alltypes:
2010 types = None
2011 data = list() # type: List[Tuple[str | List[str], ...]]
2013 resolve_mac = (
2014 self.res and isinstance(self.res[0][1].underlayer, Ether) and
2015 conf.manufdb
2016 )
2018 header = ("IP", "Service")
2019 if resolve_mac:
2020 header = ("Mac",) + header
2022 for _, r in self.res:
2023 attrs = []
2024 for attr in itertools.chain(r[DNS].an, r[DNS].ar):
2025 if types and dnstypes.get(attr.type) not in types:
2026 continue
2027 if isinstance(attr, DNSRRNSEC):
2028 attrs.append(attr.sprintf("%type%=%nextname%"))
2029 elif isinstance(attr, DNSRRSRV):
2030 attrs.append(attr.sprintf("%type%=(%target%,%port%)"))
2031 else:
2032 attrs.append(attr.sprintf("%type%=%rdata%"))
2033 ans = (r.src, attrs)
2034 if resolve_mac:
2035 mac = conf.manufdb._resolve_MAC(r.underlayer.src)
2036 data.append((mac,) + ans)
2037 else:
2038 data.append(ans)
2040 print(
2041 pretty_list(
2042 data,
2043 [header],
2044 )
2045 )
2048@conf.commands.register
2049def dnssd(service="_services._dns-sd._udp.local",
2050 af=socket.AF_INET,
2051 qtype="PTR",
2052 iface=None,
2053 verbose=2,
2054 timeout=3):
2055 """
2056 Performs a DNS-SD (RFC6763) request
2058 :param service: the service name to query (e.g. _spotify-connect._tcp.local)
2059 :param af: the transport to use. socket.AF_INET or socket.AF_INET6
2060 :param qtype: the type to use in the mDNS. Either TXT, PTR or SRV.
2061 :param iface: the interface to do this discovery on.
2063 Examples::
2065 >>> dnssd(service='_companion-link._tcp.local.').show()
2066 >>> dnssd(service='_spotify-connect._tcp.local.').show()
2067 """
2068 if af == socket.AF_INET:
2069 pkt = IP(dst=ScopedIP("224.0.0.251", iface), ttl=255)
2070 elif af == socket.AF_INET6:
2071 pkt = IPv6(dst=ScopedIP("ff02::fb", iface))
2072 else:
2073 return
2074 pkt /= UDP(sport=5353, dport=5353)
2075 pkt /= DNS(rd=0, qd=[DNSQR(qname=service, qtype=qtype)])
2076 ans, _ = sr(pkt, multi=True, timeout=timeout, verbose=verbose)
2077 return DNSSDResult(ans.res)