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